From f69e8a6c5d639138bfbb5f1efb33228ea2ab2a1a Mon Sep 17 00:00:00 2001 From: Shubham-Padkonde Date: Tue, 29 Sep 2026 22:28:55 +0530 Subject: [PATCH 1/4] feat: configure the shared RLE sequence limit --- docs/extras/vcf_annotator.md | 19 ++++++ src/ga4gh/vrs/config.py | 21 +++++++ src/ga4gh/vrs/extras/annotator/vcf.py | 6 +- src/ga4gh/vrs/extras/translator.py | 3 +- src/ga4gh/vrs/normalize.py | 5 +- tests/extras/test_rle_config.py | 90 +++++++++++++++++++++++++++ 6 files changed, 137 insertions(+), 7 deletions(-) create mode 100644 src/ga4gh/vrs/config.py create mode 100644 tests/extras/test_rle_config.py diff --git a/docs/extras/vcf_annotator.md b/docs/extras/vcf_annotator.md index eced28b0..8187a526 100644 --- a/docs/extras/vcf_annotator.md +++ b/docs/extras/vcf_annotator.md @@ -35,6 +35,25 @@ Or an alternate REST path: vrs-annotate vcf --dataproxy-uri="seqrepo+http://mylabwebsite.org/seqrepo" --vcf-out=out.vcf.gz input.vcf.gz ``` +### Configuring optional RLE sequences + +Set `GA4GH_VRS_RLE_SEQ_LIMIT` before starting Python or `vrs-annotate` to change +the maximum length of optional `ReferenceLengthExpression.sequence` values. +The default is `50` bases. Use a non-negative integer, `0` to omit the optional +sequence, or `none` (case-insensitive) for no limit. Invalid values raise a +`ValueError` when VRS-Python is imported. + +The normalizer, translator, and VCF writer share this setting. Explicit +`rle_seq_limit` arguments to normalization or translation override the default; +the VCF writer still applies the configured output limit. Required +`LiteralSequenceExpression.sequence` values are unaffected. + +For example: + +```shell +GA4GH_VRS_RLE_SEQ_LIMIT=100 vrs-annotate vcf --vrs-attributes --vcf-out=out.vcf input.vcf +``` + ### Other Options `--vrs-attributes` >Will include VRS_Start, VRS_End, VRS_State fields in the INFO field. diff --git a/src/ga4gh/vrs/config.py b/src/ga4gh/vrs/config.py new file mode 100644 index 00000000..25dd8f11 --- /dev/null +++ b/src/ga4gh/vrs/config.py @@ -0,0 +1,21 @@ +"""Process-level configuration for VRS-Python.""" + +import os + + +def _get_rle_seq_limit() -> int | None: + """Read the maximum optional RLE sequence length from the environment.""" + value = os.environ.get("GA4GH_VRS_RLE_SEQ_LIMIT", "50") + if value.lower() == "none": + return None + message = "GA4GH_VRS_RLE_SEQ_LIMIT must be a non-negative integer or 'none'" + try: + limit = int(value) + except ValueError as exc: + raise ValueError(message) from exc + if limit < 0: + raise ValueError(message) + return limit + + +RLE_SEQ_LIMIT = _get_rle_seq_limit() diff --git a/src/ga4gh/vrs/extras/annotator/vcf.py b/src/ga4gh/vrs/extras/annotator/vcf.py index 3dd70dcf..f118d758 100644 --- a/src/ga4gh/vrs/extras/annotator/vcf.py +++ b/src/ga4gh/vrs/extras/annotator/vcf.py @@ -14,6 +14,7 @@ use_ga4gh_compute_identifier_when, ) from ga4gh.vrs import VRS_VERSION, VrsType, __version__ +from ga4gh.vrs.config import RLE_SEQ_LIMIT from ga4gh.vrs.dataproxy import _DataProxy from ga4gh.vrs.extras.translator import AlleleTranslator from ga4gh.vrs.models import Allele, Range @@ -64,11 +65,6 @@ class FieldName(str, Enum): } ) -# ReferenceLengthExpression .sequence values will be included in output VCF if -# length <= this value. This field is optional for RLE since it can be derived -# from the reference sequence. Set to None to always include the sequence. -RLE_SEQ_LIMIT = 50 - def dump_alleles_to_pkl(alleles: list[Allele], output_pkl_path: Path) -> None: """Create pkl file of dictionary mapping VRS IDs to ingested alleles. diff --git a/src/ga4gh/vrs/extras/translator.py b/src/ga4gh/vrs/extras/translator.py index 91c45532..0adcdba5 100644 --- a/src/ga4gh/vrs/extras/translator.py +++ b/src/ga4gh/vrs/extras/translator.py @@ -15,6 +15,7 @@ from ga4gh.core import ga4gh_identify from ga4gh.vrs import models, normalize +from ga4gh.vrs.config import RLE_SEQ_LIMIT from ga4gh.vrs.dataproxy import SequenceProxy, _DataProxy from ga4gh.vrs.extras.decorators import lazy_property from ga4gh.vrs.normalize import denormalize_reference_length_expression @@ -74,7 +75,7 @@ def __init__( data_proxy: _DataProxy, default_assembly_name: str = "GRCh38", identify: bool = True, - rle_seq_limit: int | None = 50, + rle_seq_limit: int | None = RLE_SEQ_LIMIT, ) -> None: self.default_assembly_name = default_assembly_name self.data_proxy = data_proxy diff --git a/src/ga4gh/vrs/normalize.py b/src/ga4gh/vrs/normalize.py index fad73d87..8e3b2511 100644 --- a/src/ga4gh/vrs/normalize.py +++ b/src/ga4gh/vrs/normalize.py @@ -15,6 +15,7 @@ from ga4gh.core import ga4gh_digest, is_pydantic_instance, pydantic_copy from ga4gh.vrs import models +from ga4gh.vrs.config import RLE_SEQ_LIMIT from ga4gh.vrs.dataproxy import SequenceProxy, _DataProxy _logger = logging.getLogger(__name__) @@ -85,7 +86,9 @@ def _get_new_allele_location_pos( def _normalize_allele( - input_allele: models.Allele, data_proxy: _DataProxy, rle_seq_limit: int = 50 + input_allele: models.Allele, + data_proxy: _DataProxy, + rle_seq_limit: int | None = RLE_SEQ_LIMIT, ): """Normalize Allele using "fully-justified" normalization adapted from NCBI's VOCA. Fully-justified normalization expands such ambiguous representation over the diff --git a/tests/extras/test_rle_config.py b/tests/extras/test_rle_config.py new file mode 100644 index 00000000..bf841d41 --- /dev/null +++ b/tests/extras/test_rle_config.py @@ -0,0 +1,90 @@ +import json +import os +import subprocess +import sys + +import pytest + + +@pytest.mark.parametrize( + ("limit", "expected_sequence"), + [ + (None, "CTTTCTTT"), + ("0", None), + ("4", None), + ("8", "CTTTCTTT"), + ("none", "CTTTCTTT"), + ], +) +def test_rle_sequence_limit_environment(limit, expected_sequence, tmp_path): + """The translator and VCF writer honor the same process-level limit.""" + env = os.environ.copy() + env.pop("GA4GH_VRS_RLE_SEQ_LIMIT", None) + if limit is not None: + env["GA4GH_VRS_RLE_SEQ_LIMIT"] = limit + output = tmp_path / "annotated.vcf" + result = subprocess.run( # noqa: S603 + [ + sys.executable, + "-c", + """ +import json +import sys +import pysam +import vcr +from ga4gh.vrs.dataproxy import SeqRepoRESTDataProxy +from ga4gh.vrs import normalize +from ga4gh.vrs.extras.translator import AlleleTranslator +from ga4gh.vrs.extras.annotator.vcf import VcfAnnotator +from pathlib import Path + +with vcr.use_cassette( + 'tests/extras/cassettes/test_annotate_vcf_rle.yaml', + record_mode='none', allow_playback_repeats=True, +): + proxy = SeqRepoRESTDataProxy(base_url='http://localhost:5000/seqrepo', disable_healthcheck=True) + variant = '1-102995989-CTTT-CTTTCTTT' + state = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad').state + literal = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad', do_normalize=False) + normalized = normalize(literal, proxy).state + explicit = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad', rle_seq_limit=None).state + VcfAnnotator(proxy).annotate( + Path('tests/extras/data/test_rle.vcf'), Path(sys.argv[1]), vrs_attributes=True, + ) + with pysam.VariantFile(sys.argv[1]) as vcf: + records = list(vcf) + print(json.dumps([ + state.sequence.root if state.sequence is not None else None, + records[1].info['VRS_States'][-1], + explicit.sequence.root, + normalized.sequence.root if normalized.sequence is not None else None, + ])) +""", + str(output), + ], + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == [ + expected_sequence, + expected_sequence or ".", + "CTTTCTTT", + expected_sequence, + ] + + +@pytest.mark.parametrize("limit", ["-1", "invalid", "1.5"]) +def test_invalid_rle_sequence_limit_environment(limit): + env = dict(os.environ, GA4GH_VRS_RLE_SEQ_LIMIT=limit) + result = subprocess.run( # noqa: S603 + [sys.executable, "-c", "import ga4gh.vrs"], + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode != 0 + assert "GA4GH_VRS_RLE_SEQ_LIMIT" in result.stderr From d4ef0a2bc5f1f2a00ab0e45e95e0473962b2d5b9 Mon Sep 17 00:00:00 2001 From: Shubham-Padkonde Date: Wed, 30 Sep 2026 23:09:37 +0530 Subject: [PATCH 2/4] test: use monkeypatch for RLE environment configuration --- src/ga4gh/vrs/config.py | 11 ++- tests/extras/test_rle_config.py | 147 ++++++++++++++++---------------- 2 files changed, 84 insertions(+), 74 deletions(-) diff --git a/src/ga4gh/vrs/config.py b/src/ga4gh/vrs/config.py index 25dd8f11..d5c865fc 100644 --- a/src/ga4gh/vrs/config.py +++ b/src/ga4gh/vrs/config.py @@ -4,7 +4,16 @@ def _get_rle_seq_limit() -> int | None: - """Read the maximum optional RLE sequence length from the environment.""" + """Read the maximum optional RLE sequence length from the environment. + + Set ``GA4GH_VRS_RLE_SEQ_LIMIT`` to change the maximum length of optional + ``ReferenceLengthExpression.sequence`` values. The default is 50 bases. + Use a non-negative integer, 0 to omit the optional sequence, or ``none`` + (case-insensitive) for no limit. + + :raise ValueError: If the value is not an integer or is negative, unless + it is ``none`` (case-insensitive). + """ value = os.environ.get("GA4GH_VRS_RLE_SEQ_LIMIT", "50") if value.lower() == "none": return None diff --git a/tests/extras/test_rle_config.py b/tests/extras/test_rle_config.py index bf841d41..03b87702 100644 --- a/tests/extras/test_rle_config.py +++ b/tests/extras/test_rle_config.py @@ -1,90 +1,91 @@ -import json -import os -import subprocess -import sys +from importlib import import_module, reload +from pathlib import Path +import pysam import pytest +import vcr + +from ga4gh.vrs import config +from ga4gh.vrs.dataproxy import SeqRepoRESTDataProxy + + +@pytest.fixture +def configured_modules(monkeypatch, request): + """Reload import-time defaults, restoring them after each environment case.""" + modules = [ + config, + import_module("ga4gh.vrs.normalize"), + import_module("ga4gh.vrs.extras.translator"), + import_module("ga4gh.vrs.extras.annotator.vcf"), + ] + original_namespaces = [vars(module).copy() for module in modules] + try: + with monkeypatch.context() as env: + env.delenv("GA4GH_VRS_RLE_SEQ_LIMIT", raising=False) + if request.param is not None: + env.setenv("GA4GH_VRS_RLE_SEQ_LIMIT", request.param) + for module in modules: + reload(module) + yield modules[1:] + finally: + # Preserve class identities already imported by other test modules. + for module, namespace in zip(modules, original_namespaces, strict=True): + vars(module).clear() + vars(module).update(namespace) @pytest.mark.parametrize( - ("limit", "expected_sequence"), + ("configured_modules", "expected_sequence"), [ (None, "CTTTCTTT"), ("0", None), ("4", None), ("8", "CTTTCTTT"), ("none", "CTTTCTTT"), + ("NoNe", "CTTTCTTT"), ], + indirect=["configured_modules"], ) -def test_rle_sequence_limit_environment(limit, expected_sequence, tmp_path): - """The translator and VCF writer honor the same process-level limit.""" - env = os.environ.copy() - env.pop("GA4GH_VRS_RLE_SEQ_LIMIT", None) - if limit is not None: - env["GA4GH_VRS_RLE_SEQ_LIMIT"] = limit - output = tmp_path / "annotated.vcf" - result = subprocess.run( # noqa: S603 - [ - sys.executable, - "-c", - """ -import json -import sys -import pysam -import vcr -from ga4gh.vrs.dataproxy import SeqRepoRESTDataProxy -from ga4gh.vrs import normalize -from ga4gh.vrs.extras.translator import AlleleTranslator -from ga4gh.vrs.extras.annotator.vcf import VcfAnnotator -from pathlib import Path - -with vcr.use_cassette( - 'tests/extras/cassettes/test_annotate_vcf_rle.yaml', - record_mode='none', allow_playback_repeats=True, +def test_rle_sequence_limit_environment( + configured_modules, expected_sequence, tmp_path ): - proxy = SeqRepoRESTDataProxy(base_url='http://localhost:5000/seqrepo', disable_healthcheck=True) - variant = '1-102995989-CTTT-CTTTCTTT' - state = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad').state - literal = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad', do_normalize=False) - normalized = normalize(literal, proxy).state - explicit = AlleleTranslator(proxy).translate_from(variant, fmt='gnomad', rle_seq_limit=None).state - VcfAnnotator(proxy).annotate( - Path('tests/extras/data/test_rle.vcf'), Path(sys.argv[1]), vrs_attributes=True, - ) - with pysam.VariantFile(sys.argv[1]) as vcf: - records = list(vcf) - print(json.dumps([ - state.sequence.root if state.sequence is not None else None, - records[1].info['VRS_States'][-1], - explicit.sequence.root, - normalized.sequence.root if normalized.sequence is not None else None, - ])) -""", - str(output), - ], - env=env, - capture_output=True, - text=True, - check=False, - ) - assert result.returncode == 0, result.stderr - assert json.loads(result.stdout) == [ - expected_sequence, - expected_sequence or ".", - "CTTTCTTT", - expected_sequence, - ] + """Normalization, translation and VCF output share the configured default.""" + normalization, translator, vcf_module = configured_modules + output = tmp_path / "annotated.vcf" + with vcr.use_cassette( + "tests/extras/cassettes/test_annotate_vcf_rle.yaml", + record_mode="none", + allow_playback_repeats=True, + ): + proxy = SeqRepoRESTDataProxy( + base_url="http://localhost:5000/seqrepo", + disable_healthcheck=True, + ) + tlr = translator.AlleleTranslator(proxy) + variant = "1-102995989-CTTT-CTTTCTTT" + state = tlr.translate_from(variant, fmt="gnomad").state + assert ( + state.sequence.root if state.sequence is not None else None + ) == expected_sequence + literal = tlr.translate_from(variant, fmt="gnomad", do_normalize=False) + normalized = normalization.normalize(literal, proxy).state + assert ( + normalized.sequence.root if normalized.sequence is not None else None + ) == expected_sequence + explicit = tlr.translate_from(variant, fmt="gnomad", rle_seq_limit=None).state + assert explicit.sequence.root == "CTTTCTTT" + vcf_module.VcfAnnotator(proxy).annotate( + Path("tests/extras/data/test_rle.vcf"), + output, + vrs_attributes=True, + ) + with pysam.VariantFile(output) as vcf: + records = list(vcf) + assert records[1].info["VRS_States"][-1] == (expected_sequence or ".") @pytest.mark.parametrize("limit", ["-1", "invalid", "1.5"]) -def test_invalid_rle_sequence_limit_environment(limit): - env = dict(os.environ, GA4GH_VRS_RLE_SEQ_LIMIT=limit) - result = subprocess.run( # noqa: S603 - [sys.executable, "-c", "import ga4gh.vrs"], - env=env, - capture_output=True, - text=True, - check=False, - ) - assert result.returncode != 0 - assert "GA4GH_VRS_RLE_SEQ_LIMIT" in result.stderr +def test_invalid_rle_sequence_limit_environment(limit, monkeypatch): + monkeypatch.setenv("GA4GH_VRS_RLE_SEQ_LIMIT", limit) + with pytest.raises(ValueError, match="GA4GH_VRS_RLE_SEQ_LIMIT"): + config._get_rle_seq_limit() From 806445e1604bf33273278cb4aa2b09c6d7c85feb Mon Sep 17 00:00:00 2001 From: Shubham Padkonde Date: Fri, 16 Oct 2026 16:30:10 +0530 Subject: [PATCH 3/4] test: cover configured RLE limits in existing workflows --- tests/extras/conftest.py | 30 ++++++++++ tests/extras/test_allele_translator.py | 27 ++++++++- tests/extras/test_annotate_vcf.py | 22 ++++++- tests/extras/test_rle_config.py | 81 -------------------------- 4 files changed, 75 insertions(+), 85 deletions(-) create mode 100644 tests/extras/conftest.py diff --git a/tests/extras/conftest.py b/tests/extras/conftest.py new file mode 100644 index 00000000..ed33ec2e --- /dev/null +++ b/tests/extras/conftest.py @@ -0,0 +1,30 @@ +from importlib import import_module, reload + +import pytest + +from ga4gh.vrs import config + + +@pytest.fixture +def configured_modules(monkeypatch, request): + """Reload import-time defaults, restoring them after each environment case.""" + modules = [ + config, + import_module("ga4gh.vrs.normalize"), + import_module("ga4gh.vrs.extras.translator"), + import_module("ga4gh.vrs.extras.annotator.vcf"), + ] + original_namespaces = [vars(module).copy() for module in modules] + try: + with monkeypatch.context() as env: + env.delenv("GA4GH_VRS_RLE_SEQ_LIMIT", raising=False) + if request.param is not None: + env.setenv("GA4GH_VRS_RLE_SEQ_LIMIT", request.param) + for module in modules: + reload(module) + yield modules[1:] + finally: + # Preserve class identities already imported by other test modules. + for module, namespace in zip(modules, original_namespaces, strict=True): + vars(module).clear() + vars(module).update(namespace) diff --git a/tests/extras/test_allele_translator.py b/tests/extras/test_allele_translator.py index 15f87a89..6f4a95ef 100644 --- a/tests/extras/test_allele_translator.py +++ b/tests/extras/test_allele_translator.py @@ -307,8 +307,24 @@ def tlr(rest_dataproxy): @pytest.mark.vcr -def test_rle_round_trip_gnomad_spdi(tlr): +@pytest.mark.parametrize( + ("configured_modules", "limit", "vcr_cassette_name"), + [ + (None, 50, "test_rle_round_trip_gnomad_spdi"), + ("0", 0, "test_rle_round_trip_gnomad_spdi"), + ("1", 1, "test_rle_round_trip_gnomad_spdi"), + ("2", 2, "test_rle_round_trip_gnomad_spdi"), + ("none", None, "test_rle_round_trip_gnomad_spdi"), + ("NoNe", None, "test_rle_round_trip_gnomad_spdi"), + ], + indirect=["configured_modules"], +) +def test_rle_round_trip_gnomad_spdi( + configured_modules, limit, rest_dataproxy, vcr_cassette +): """Test translating an RLE allele from gnomAD and SPDI, and back to SPDI""" + tlr = configured_modules[1].AlleleTranslator(rest_dataproxy, identify=False) + vcr_cassette.allow_playback_repeats = True for inp in rle_inputs: gnomad = inp["gnomad"] spdi = inp["spdi"] @@ -318,6 +334,15 @@ def test_rle_round_trip_gnomad_spdi(tlr): allele_spdi = tlr.translate_from(spdi, fmt="spdi", rle_seq_limit=None) assert allele_spdi.model_dump(exclude_none=True) == expected_dict + # Per-call None above includes the sequence regardless of the environment. + # Omitting the argument uses the configured default in both formats. + expected_default = {**expected_dict, "state": expected_dict["state"].copy()} + if limit is not None and expected_default["state"]["length"] > limit: + expected_default["state"].pop("sequence", None) + for value, fmt in ((gnomad, "gnomad"), (spdi, "spdi")): + default_allele = tlr.translate_from(value, fmt=fmt) + assert default_allele.model_dump(exclude_none=True) == expected_default + allele_gnomad_to_spdi = tlr.translate_to( allele_gnomad, fmt="spdi", ref_seq_limit=None ) diff --git a/tests/extras/test_annotate_vcf.py b/tests/extras/test_annotate_vcf.py index 0f5ca0e6..43307aa2 100644 --- a/tests/extras/test_annotate_vcf.py +++ b/tests/extras/test_annotate_vcf.py @@ -233,17 +233,32 @@ def test_get_vrs_object_invalid_input(vcf_annotator: VcfAnnotator, caplog): @pytest.mark.vcr -def test_annotate_vcf_rle(vcf_annotator: VcfAnnotator, vcr_cassette): +@pytest.mark.parametrize( + ("configured_modules", "expected_sequence", "vcr_cassette_name"), + [ + (None, "CTTTCTTT", "test_annotate_vcf_rle"), + ("0", None, "test_annotate_vcf_rle"), + ("4", None, "test_annotate_vcf_rle"), + ("8", "CTTTCTTT", "test_annotate_vcf_rle"), + ("none", "CTTTCTTT", "test_annotate_vcf_rle"), + ("NoNe", "CTTTCTTT", "test_annotate_vcf_rle"), + ], + indirect=["configured_modules"], +) +def test_annotate_vcf_rle( + configured_modules, expected_sequence, rest_dataproxy_fn_scope, vcr_cassette, tmp_path +): """Test VCF annotation with ReferenceLengthExpression states. Tests two variants that should produce ReferenceLengthExpression states: 1. Deletion (1:100210778 AA>A): Expected RLE with length=1, repeatSubunitLength=1 2. Duplication (1:102995989 CTTT>CTTTCTTT): Expected RLE with length=8, repeatSubunitLength=4 """ + vcf_annotator = configured_modules[2].VcfAnnotator(rest_dataproxy_fn_scope) vcr_cassette.allow_playback_repeats = False input_vcf = TEST_DATA_DIR / "test_rle.vcf" - output_vcf = TEST_DATA_DIR / "test_rle_output.vcf" - output_vrs_pkl = TEST_DATA_DIR / "test_rle_output.pkl" + output_vcf = tmp_path / "test_rle_output.vcf" + output_vrs_pkl = tmp_path / "test_rle_output.pkl" # Annotate the VCF with VRS attributes enabled vcf_annotator.annotate( @@ -295,6 +310,7 @@ def test_annotate_vcf_rle(vcf_annotator: VcfAnnotator, vcr_cassette): assert duplication_variant.pos == 102995989 assert duplication_variant.ref == "CTTT" assert duplication_variant.alts == ("CTTTCTTT",) + assert duplication_variant.info["VRS_States"][-1] == (expected_sequence or ".") # Check VRS attributes for duplication assert "VRS_Allele_IDs" in duplication_variant.info diff --git a/tests/extras/test_rle_config.py b/tests/extras/test_rle_config.py index 03b87702..8b17c41f 100644 --- a/tests/extras/test_rle_config.py +++ b/tests/extras/test_rle_config.py @@ -1,87 +1,6 @@ -from importlib import import_module, reload -from pathlib import Path - -import pysam import pytest -import vcr from ga4gh.vrs import config -from ga4gh.vrs.dataproxy import SeqRepoRESTDataProxy - - -@pytest.fixture -def configured_modules(monkeypatch, request): - """Reload import-time defaults, restoring them after each environment case.""" - modules = [ - config, - import_module("ga4gh.vrs.normalize"), - import_module("ga4gh.vrs.extras.translator"), - import_module("ga4gh.vrs.extras.annotator.vcf"), - ] - original_namespaces = [vars(module).copy() for module in modules] - try: - with monkeypatch.context() as env: - env.delenv("GA4GH_VRS_RLE_SEQ_LIMIT", raising=False) - if request.param is not None: - env.setenv("GA4GH_VRS_RLE_SEQ_LIMIT", request.param) - for module in modules: - reload(module) - yield modules[1:] - finally: - # Preserve class identities already imported by other test modules. - for module, namespace in zip(modules, original_namespaces, strict=True): - vars(module).clear() - vars(module).update(namespace) - - -@pytest.mark.parametrize( - ("configured_modules", "expected_sequence"), - [ - (None, "CTTTCTTT"), - ("0", None), - ("4", None), - ("8", "CTTTCTTT"), - ("none", "CTTTCTTT"), - ("NoNe", "CTTTCTTT"), - ], - indirect=["configured_modules"], -) -def test_rle_sequence_limit_environment( - configured_modules, expected_sequence, tmp_path -): - """Normalization, translation and VCF output share the configured default.""" - normalization, translator, vcf_module = configured_modules - output = tmp_path / "annotated.vcf" - with vcr.use_cassette( - "tests/extras/cassettes/test_annotate_vcf_rle.yaml", - record_mode="none", - allow_playback_repeats=True, - ): - proxy = SeqRepoRESTDataProxy( - base_url="http://localhost:5000/seqrepo", - disable_healthcheck=True, - ) - tlr = translator.AlleleTranslator(proxy) - variant = "1-102995989-CTTT-CTTTCTTT" - state = tlr.translate_from(variant, fmt="gnomad").state - assert ( - state.sequence.root if state.sequence is not None else None - ) == expected_sequence - literal = tlr.translate_from(variant, fmt="gnomad", do_normalize=False) - normalized = normalization.normalize(literal, proxy).state - assert ( - normalized.sequence.root if normalized.sequence is not None else None - ) == expected_sequence - explicit = tlr.translate_from(variant, fmt="gnomad", rle_seq_limit=None).state - assert explicit.sequence.root == "CTTTCTTT" - vcf_module.VcfAnnotator(proxy).annotate( - Path("tests/extras/data/test_rle.vcf"), - output, - vrs_attributes=True, - ) - with pysam.VariantFile(output) as vcf: - records = list(vcf) - assert records[1].info["VRS_States"][-1] == (expected_sequence or ".") @pytest.mark.parametrize("limit", ["-1", "invalid", "1.5"]) From 94323478b3a0030de0096d5511f7c5913e3b7f9f Mon Sep 17 00:00:00 2001 From: Shubham-Padkonde Date: Fri, 16 Oct 2026 16:31:10 +0530 Subject: [PATCH 4/4] style: format parameterized annotation test --- tests/extras/test_annotate_vcf.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/extras/test_annotate_vcf.py b/tests/extras/test_annotate_vcf.py index 43307aa2..0d749bd4 100644 --- a/tests/extras/test_annotate_vcf.py +++ b/tests/extras/test_annotate_vcf.py @@ -246,7 +246,11 @@ def test_get_vrs_object_invalid_input(vcf_annotator: VcfAnnotator, caplog): indirect=["configured_modules"], ) def test_annotate_vcf_rle( - configured_modules, expected_sequence, rest_dataproxy_fn_scope, vcr_cassette, tmp_path + configured_modules, + expected_sequence, + rest_dataproxy_fn_scope, + vcr_cassette, + tmp_path, ): """Test VCF annotation with ReferenceLengthExpression states.