Unit test for new variant CSV download with post-mapped HGVS

Author: jstone-dev

tests/helpers/constants.py

@@ -987,6 +987,15 @@
     }
 }
 
+TEST_POST_MAPPED_METADATA_WITH_EXPRESSION = {
+    "genomic": {
+        "sequence_id": "ga4gh:SQ.em9khDCUYXrVWBfWr9r8fjBUrTjj1aig",
+        "sequence_type": "dna",
+        "sequence_accessions": [VALID_NT_ACCESSION],
+        "sequence_genes": [VALID_GENE],
+    }
+}
+
 TEST_SEQ_SCORESET_VARIANT_MAPPING_SCAFFOLD = {
     "metadata": {},
     "reference_sequences": {
@@ -1083,6 +1092,86 @@
     "mapping_api_version": "pytest.0.0",
 }
 
+TEST_POST_MAPPED_VRS_WITH_HGVS_G_EXPRESSION = {
+    "id": "ga4gh:VA.fRW7u-kBQnAKitu1PoDMLvlECWZTHCos",
+    "type": "Allele",
+    "state": {
+        "type": "LiteralSequenceExpression",
+        "sequence": "G"
+    },
+    "digest": "fRW7u-kBQnAKitu1PoDMLvlECWZTHCos",
+    "location": {
+        "id": "ga4gh:SL.99b3WBaSSmaSTs6YmJfIhl1ZDCV07VZY",
+        "end": 23536836,
+        "type": "SequenceLocation",
+        "start": 23536835,
+        "digest": "99b3WBaSSmaSTs6YmJfIhl1ZDCV07VZY",
+        "sequenceReference": {
+            "type": "SequenceReference",
+            "label": "NC_000018.10",
+            "refgetAccession": "SQ.vWwFhJ5lQDMhh-czg06YtlWqu0lvFAZV"
+        }
+    },
+    "extensions": [{
+        "name": "vrs_ref_allele_seq",
+        "type": "Extension",
+        "value": "C"
+    }],
+    "expressions": [{
+        "value": "NC_000018.10:g.23536836C>G",
+        "syntax": "hgvs.g"
+    }]
+}
+
+TEST_POST_MAPPED_VRS_WITH_HGVS_P_EXPRESSION = {
+    "id": "ga4gh:VA.zkOAzZK5qG0D0mkJUfXlK1aS075OGSjh",
+    "type": "Allele",
+    "state": {
+        "type": "LiteralSequenceExpression",
+        "sequence": "R"
+    },
+    "digest": "zkOAzZK5qG0D0mkJUfXlK1aS075OGSjh",
+    "location": {
+        "id": "ga4gh:SL.uUyRpJbrPttRThL7A2zeWAnTcb_7f1R2",
+        "end": 116,
+        "type": "SequenceLocation",
+        "start": 115,
+        "digest": "uUyRpJbrPttRThL7A2zeWAnTcb_7f1R2",
+        "sequenceReference": {
+            "type": "SequenceReference",
+            "refgetAccession": "SQ.StlJo3M4b8cS253ufe9nPpWqQHBDOSPs"
+        }
+    },
+    "extensions": [{
+        "name": "vrs_ref_allele_seq",
+        "type": "Extension",
+        "value": "Q"
+    }],
+    "expressions": [{
+        "value": "NP_002746.1:p.Gln116Arg",
+        "syntax": "hgvs.p"
+    }]
+}
+
+TEST_MAPPED_VARIANT_WITH_HGVS_G_EXPRESSION = {
+    "pre_mapped": {},
+    "post_mapped": TEST_POST_MAPPED_VRS_WITH_HGVS_G_EXPRESSION,
+    "modification_date": datetime.isoformat(datetime.now()),
+    "mapped_date": datetime.isoformat(datetime.now()),
+    "current": True,
+    "vrs_version": "2.0",
+    "mapping_api_version": "pytest.0.0",
+}
+
+TEST_MAPPED_VARIANT_WITH_HGVS_P_EXPRESSION = {
+    "pre_mapped": {},
+    "post_mapped": TEST_POST_MAPPED_VRS_WITH_HGVS_P_EXPRESSION,
+    "modification_date": datetime.isoformat(datetime.now()),
+    "mapped_date": datetime.isoformat(datetime.now()),
+    "current": True,
+    "vrs_version": "2.0",
+    "mapping_api_version": "pytest.0.0",
+}
 
 TEST_WT_SCORE = 1.0
 

tests/helpers/util/variant.py

@@ -17,7 +17,6 @@
 from mavedb.models.variant import Variant
 
 from tests.helpers.constants import (
-    TEST_MINIMAL_MAPPED_VARIANT,
     TEST_MINIMAL_PRE_MAPPED_METADATA,
     TEST_MINIMAL_POST_MAPPED_METADATA,
 )
@@ -84,13 +83,13 @@ def mock_worker_variant_insertion(
     return client.get(f"api/v1/score-sets/{score_set['urn']}").json()
 
 
-def create_mapped_variants_for_score_set(db, score_set_urn):
+def create_mapped_variants_for_score_set(db, score_set_urn, mapped_variant: dict[str, any]):
     score_set = db.scalar(select(ScoreSet).where(ScoreSet.urn == score_set_urn))
     targets = db.scalars(select(TargetGene).where(TargetGene.score_set_id == score_set.id))
     variants = db.scalars(select(Variant).where(Variant.score_set_id == score_set.id)).all()
 
     for variant in variants:
-        mv = MappedVariant(**TEST_MINIMAL_MAPPED_VARIANT, variant_id=variant.id)
+        mv = MappedVariant(**mapped_variant, variant_id=variant.id)
         db.add(mv)
 
     for target in targets:

tests/routers/test_score_set.py

@@ -2,7 +2,9 @@
 
 import re
 from copy import deepcopy
+import csv
 from datetime import date
+from io import StringIO
 from unittest.mock import patch
 
 import jsonschema
@@ -27,6 +29,8 @@
     EXTRA_USER,
     EXTRA_LICENSE,
     TEST_CROSSREF_IDENTIFIER,
+    TEST_MAPPED_VARIANT_WITH_HGVS_G_EXPRESSION,
+    TEST_MAPPED_VARIANT_WITH_HGVS_P_EXPRESSION,
     TEST_MINIMAL_ACC_SCORESET,
     TEST_MINIMAL_SEQ_SCORESET,
     TEST_MINIMAL_SEQ_SCORESET_RESPONSE,
@@ -62,7 +66,7 @@
     publish_score_set,
 )
 from tests.helpers.util.user import change_ownership
-from tests.helpers.util.variant import mock_worker_variant_insertion
+from tests.helpers.util.variant import create_mapped_variants_for_score_set, mock_worker_variant_insertion
 
 
 ########################################################################################################################
@@ -2347,6 +2351,44 @@ def test_upload_a_non_utf8_file(session, client, setup_router_db, data_files):
 # Score set download files
 ########################################################################################################################
 
+@pytest.mark.parametrize("mapped_variant,has_hgvs_g,has_hgvs_p", [(None, False, False), (TEST_MAPPED_VARIANT_WITH_HGVS_G_EXPRESSION, True, False), (TEST_MAPPED_VARIANT_WITH_HGVS_P_EXPRESSION, False, True)], ids=["without_post_mapped_vrs", "with_post_mapped_hgvs_g", "with_post_mapped_hgvs_p"])
+def test_download_variants_data_file(session, data_provider, client, setup_router_db, data_files, mapped_variant, has_hgvs_g, has_hgvs_p):
+    experiment = create_experiment(client)
+    score_set = create_seq_score_set(client, experiment["urn"])
+    score_set = mock_worker_variant_insertion(client, session, data_provider, score_set, data_files / "scores.csv")
+    if mapped_variant is not None:
+        create_mapped_variants_for_score_set(session, score_set["urn"], mapped_variant)
+
+    with patch.object(arq.ArqRedis, "enqueue_job", return_value=None) as worker_queue:
+        published_score_set = publish_score_set(client, score_set["urn"])
+        worker_queue.assert_called_once()
+
+    download_scores_csv_response = client.get(
+        f"/api/v1/score-sets/{published_score_set['urn']}/variants/data?drop_na_columns=true"
+    )
+    assert download_scores_csv_response.status_code == 200
+    download_scores_csv = download_scores_csv_response.text
+
+    reader = csv.DictReader(StringIO(download_scores_csv))
+    assert sorted(reader.fieldnames) == sorted([
+        "accession",
+        "hgvs_nt",
+        "hgvs_pro",
+        "post_mapped_hgvs_g",
+        "post_mapped_hgvs_p",
+        "score",
+    ])
+    rows = list(reader)
+    for row in rows:
+        if has_hgvs_g:
+            assert row["post_mapped_hgvs_g"] == mapped_variant["post_mapped"]["expressions"][0]["value"]
+        else:
+            assert row["post_mapped_hgvs_g"] == "NA"
+        if has_hgvs_p:
+            assert row["post_mapped_hgvs_p"] == mapped_variant["post_mapped"]["expressions"][0]["value"]
+        else:
+            assert row["post_mapped_hgvs_p"] == "NA"
+
 
 # Test file doesn't have hgvs_splice so its values are all NA.
 def test_download_scores_file(session, data_provider, client, setup_router_db, data_files):

tests/routers/test_statistics.py

@@ -12,6 +12,7 @@
 
 from tests.helpers.constants import (
     TEST_BIORXIV_IDENTIFIER,
+    TEST_MINIMAL_MAPPED_VARIANT,
     TEST_NT_CDOT_TRANSCRIPT,
     TEST_KEYWORDS,
     TEST_MEDRXIV_IDENTIFIER,
@@ -64,7 +65,7 @@ def setup_seq_scoreset(setup_router_db, session, data_provider, client, data_fil
     unpublished_score_set = mock_worker_variant_insertion(
         client, session, data_provider, unpublished_score_set, data_files / "scores.csv"
     )
-    create_mapped_variants_for_score_set(session, unpublished_score_set["urn"])
+    create_mapped_variants_for_score_set(session, unpublished_score_set["urn"], TEST_MINIMAL_MAPPED_VARIANT)
 
     with patch.object(arq.ArqRedis, "enqueue_job", return_value=None) as worker_queue:
         publish_score_set(client, unpublished_score_set["urn"])