From 22f2299c44385fdffd96caaa88927527bf9f4c83 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Mon, 10 Aug 2026 17:23:51 -0700 Subject: [PATCH 1/7] fix(api): return uncaught-exception 500s through the CORS layer Starlette dispatches @app.exception_handler(Exception) from ServerErrorMiddleware, which is installed outside the user middleware stack. Its 500 never passes through CORSMiddleware and so carries no Access-Control-Allow-Origin; the browser rejects it before axios sees the body, and the caller gets an opaque network error instead of an attributable failure. Catch the exception in a pure ASGI middleware installed as the innermost layer, so CORS decorates the response and the correlation id is available to put in the body. Pure ASGI rather than BaseHTTPMiddleware so the NDJSON streaming endpoints are left alone; exceptions raised after the response starts are re-raised, since the status is already committed. The response body uses 'detail' to match every other error path in the app. The handler in server_main stays as a backstop for anything raised outside the middleware. --- src/mavedb/lib/middleware/__init__.py | 7 +++ src/mavedb/lib/middleware/errors.py | 75 +++++++++++++++++++++++ src/mavedb/server_main.py | 5 ++ tests/lib/middleware/__init__.py | 0 tests/lib/middleware/test_errors.py | 87 +++++++++++++++++++++++++++ 5 files changed, 174 insertions(+) create mode 100644 src/mavedb/lib/middleware/__init__.py create mode 100644 src/mavedb/lib/middleware/errors.py create mode 100644 tests/lib/middleware/__init__.py create mode 100644 tests/lib/middleware/test_errors.py diff --git a/src/mavedb/lib/middleware/__init__.py b/src/mavedb/lib/middleware/__init__.py new file mode 100644 index 000000000..04e9e2b9f --- /dev/null +++ b/src/mavedb/lib/middleware/__init__.py @@ -0,0 +1,7 @@ +"""ASGI middleware for the MaveDB application.""" + +from mavedb.lib.middleware.errors import CatchAllErrorMiddleware + +__all__ = [ + "CatchAllErrorMiddleware", +] diff --git a/src/mavedb/lib/middleware/errors.py b/src/mavedb/lib/middleware/errors.py new file mode 100644 index 000000000..871eac5d9 --- /dev/null +++ b/src/mavedb/lib/middleware/errors.py @@ -0,0 +1,75 @@ +"""Middleware that converts an uncaught exception into a response the caller can read.""" + +import logging +import time + +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from mavedb.lib.logging.canonical import log_request +from mavedb.lib.logging.context import ( + correlation_id_for_context, + format_raised_exception_info_as_dict, + logging_context, + save_to_logging_context, +) +from mavedb.lib.slack import send_slack_error + +logger = logging.getLogger(__name__) + + +class CatchAllErrorMiddleware: + """Turn an uncaught exception into a 500 that the browser is allowed to read. + + Starlette dispatches ``@app.exception_handler(Exception)`` from ``ServerErrorMiddleware``, which sits + outside the user middleware stack. Its response therefore never passes through ``CORSMiddleware`` and + carries no ``Access-Control-Allow-Origin``, so a browser rejects it before the client library sees the + body and the caller is left with an opaque network error. Installing this middleware *inside* the CORS + layer puts the 500 back under CORS, and inside the context middleware so the correlation id that + identifies the failure in the logs can be returned to the caller. + + Implemented as pure ASGI rather than ``BaseHTTPMiddleware``: the latter interposes on the response body + and has a history of breaking ``StreamingResponse``, which the NDJSON export endpoints rely on. Only + exceptions raised before the response starts are converted — once bytes are on the wire the status is + already committed, so the exception is re-raised and each stream is responsible for its own + mid-flight error reporting. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + response_started = False + + async def send_wrapper(message: Message) -> None: + nonlocal response_started + if message["type"] == "http.response.start": + response_started = True + await send(message) + + try: + await self.app(scope, receive, send_wrapper) + except Exception as err: + # The status line is already committed; nothing can be salvaged into a 500 here. + if response_started: + raise + + save_to_logging_context(format_raised_exception_info_as_dict(err)) + request = Request(scope, receive) + response = JSONResponse( + status_code=500, + content={"detail": "Internal server error", "correlation_id": correlation_id_for_context()}, + ) + + try: + logger.error(msg="Uncaught exception.", extra=logging_context(), exc_info=err) + send_slack_error(err=err, request=request) + finally: + log_request(request, response, time.time_ns()) + + await response(scope, receive, send) diff --git a/src/mavedb/server_main.py b/src/mavedb/server_main.py index 150714883..880bfcfe1 100644 --- a/src/mavedb/server_main.py +++ b/src/mavedb/server_main.py @@ -34,6 +34,7 @@ logging_context, save_to_logging_context, ) +from mavedb.lib.middleware import CatchAllErrorMiddleware from mavedb.lib.permissions.exceptions import PermissionException from mavedb.lib.slack import send_slack_error from mavedb.models import * # noqa: F403 @@ -75,6 +76,10 @@ configure_mappers() app = FastAPI() +# `add_middleware` inserts at the head of the stack, so the *first* call here is the innermost layer. +# CatchAllErrorMiddleware must sit inside both CORSMiddleware and the context middleware: CORS has to +# decorate the 500 it produces, and the correlation id it returns comes from the context. +app.add_middleware(CatchAllErrorMiddleware) app.add_middleware( PopulatedRawContextMiddleware, plugins=( diff --git a/tests/lib/middleware/__init__.py b/tests/lib/middleware/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/lib/middleware/test_errors.py b/tests/lib/middleware/test_errors.py new file mode 100644 index 000000000..76a14b0e2 --- /dev/null +++ b/tests/lib/middleware/test_errors.py @@ -0,0 +1,87 @@ +"""Tests for the application middleware stack. + +These assert the *wiring* in ``mavedb.server_main``, not the middleware class in isolation: the defect +being guarded against is an ordering mistake, which only a test against the real stack can catch. +""" + +# ruff: noqa: E402 + +import logging +from unittest.mock import patch + +import pytest + +pytest.importorskip("psycopg2") +pytest.importorskip("fastapi") + +from fastapi.testclient import TestClient + +from mavedb.server_main import app + +BOOM_PATH = "/__test_boom_!__" +TEST_ORIGIN = "https://www.mavedb.org" + + +class DeliberateFailure(Exception): + """Raised only by the probe route below.""" + + +@pytest.fixture +def raising_route(): + """Register a route that raises, and take it back out again afterwards.""" + + @app.get(BOOM_PATH) + def boom(): + raise DeliberateFailure("oh no!") + + route = app.router.routes[-1] + try: + yield + finally: + app.router.routes.remove(route) + + +@pytest.fixture +def slack_error(): + with patch("mavedb.lib.middleware.errors.send_slack_error") as mocked: + yield mocked + + +@pytest.mark.unit +class TestCatchAllErrorMiddleware: + def test_uncaught_exception_returns_500_with_cors_headers(self, raising_route, slack_error): + """A browser can only read the error if the 500 passed through CORSMiddleware.""" + with TestClient(app) as tc: + response = tc.get(BOOM_PATH, headers={"Origin": TEST_ORIGIN}) + + assert response.status_code == 500 + assert response.headers.get("access-control-allow-origin") in ("*", TEST_ORIGIN) + + def test_uncaught_exception_body_is_attributable(self, raising_route, slack_error): + with TestClient(app) as tc: + response = tc.get(BOOM_PATH, headers={"Origin": TEST_ORIGIN}) + + body = response.json() + assert body["detail"] == "Internal server error" + assert body["correlation_id"] + + def test_uncaught_exception_still_alerts_slack(self, raising_route, slack_error): + with TestClient(app) as tc: + tc.get(BOOM_PATH) + + slack_error.assert_called_once() + assert isinstance(slack_error.call_args.kwargs["err"], DeliberateFailure) + + def test_uncaught_exception_still_logs(self, raising_route, slack_error, caplog): + with caplog.at_level(logging.ERROR), TestClient(app) as tc: + tc.get(BOOM_PATH) + + assert any("Uncaught exception." in record.message for record in caplog.records) + + def test_successful_request_is_untouched(self, slack_error): + with TestClient(app) as tc: + response = tc.get("/api/v1/api/version", headers={"Origin": TEST_ORIGIN}) + + assert response.status_code == 200 + assert response.headers.get("access-control-allow-origin") in ("*", TEST_ORIGIN) + slack_error.assert_not_called() From e1436cff6e032a3456a454b7e927dda14cf29186 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Mon, 10 Aug 2026 18:02:36 -0700 Subject: [PATCH 2/7] fix(api): report a failed variant in-band rather than truncating the stream _stream_generated_annotations caught only MappingDataDoesntExistException, so anything else raised after the first yield ended the body mid-stream. The 200 and its headers went out with the first chunk, leaving the consumer a short file it cannot distinguish from a complete one. Three known raising paths reach it: absent or malformed score data, an unrecognized VRS Allele state type, and an empty post-mapped id list. Classify each variant instead. A failure becomes a record carrying an error object, and serialization is inside the try -- an emitted object that builds and then fails to dump is the same shape-dependent failure, and is what 5c155f4d fixed. A missing mapping stays an expected null so consumers can keep telling the two apart. Outcome counts go to the logs rather than a trailing summary record: every line stays a variant record, the body still holds exactly X-Total-Count lines, and the format stays identical to the public dump's va/{urn}.va.ndjson. The contract change is additive. --- src/mavedb/routers/score_sets.py | 148 +++++++++++++++++++++------- tests/routers/test_score_set.py | 163 +++++++++++++++++++++++++++++++ 2 files changed, 273 insertions(+), 38 deletions(-) diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index de0cda700..d74a0abba 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -4,7 +4,7 @@ import time from datetime import date, datetime from functools import partial -from typing import Any, List, Optional, Sequence, TypedDict, Union +from typing import Any, List, Literal, Optional, Sequence, TypedDict, Union import numpy as np import pandas as pd @@ -36,6 +36,15 @@ require_current_user_with_email, ) from mavedb.lib.contributors import find_or_create_contributor +from mavedb.lib.csv.columns import variants_to_csv_rows +from mavedb.lib.csv.deprecated_params import ( + DROP_NA_COLUMNS_DESCRIPTION, + INCLUDE_CUSTOM_COLUMNS_DESCRIPTION, + INCLUDE_POST_MAPPED_HGVS_DESCRIPTION, + resolve_deprecated_csv_params, +) +from mavedb.lib.csv.namespaces import CSV_NAMESPACES_PARAM_DESCRIPTION, CsvNamespaceStr +from mavedb.lib.csv.score_set import available_score_set_csv_namespaces, get_score_set_variants_as_csv from mavedb.lib.exceptions import MixedTargetError, NonexistentOrcidUserError from mavedb.lib.experiments import enrich_experiment_with_num_score_sets from mavedb.lib.identifiers import ( @@ -53,16 +62,6 @@ from mavedb.lib.permissions.principal import Principal from mavedb.lib.permissions.score_calibration import ScoreCalibrationViewer from mavedb.lib.score_calibrations import create_score_calibration -from mavedb.lib.csv.deprecated_params import ( - DROP_NA_COLUMNS_DESCRIPTION, - INCLUDE_CUSTOM_COLUMNS_DESCRIPTION, - INCLUDE_POST_MAPPED_HGVS_DESCRIPTION, - resolve_deprecated_csv_params, -) -from mavedb.lib.csv.namespaces import CSV_NAMESPACES_PARAM_DESCRIPTION, CsvNamespaceStr -from mavedb.view_models.csv_namespace import AvailableCsvNamespace -from mavedb.lib.csv.columns import variants_to_csv_rows -from mavedb.lib.csv.score_set import available_score_set_csv_namespaces, get_score_set_variants_as_csv from mavedb.lib.score_sets import ( csv_data_to_df, fetch_score_set_search_filter_options, @@ -108,6 +107,7 @@ ) from mavedb.view_models import clinical_control, gnomad_variant, mapped_variant, score_set from mavedb.view_models.contributor import ContributorCreate +from mavedb.view_models.csv_namespace import AvailableCsvNamespace from mavedb.view_models.doi_identifier import DoiIdentifierCreate from mavedb.view_models.publication_identifier import PublicationIdentifierCreate from mavedb.view_models.score_set_dataset_columns import DatasetColumnMetadata @@ -1268,6 +1268,46 @@ def get_score_set_mapped_variants( return mapped_variants +def _annotation_stream_record( + mapped_variant, annotation_function +) -> tuple[dict, Literal["annotated", "unannotated", "errored"]]: + """ + Build the NDJSON record for one mapped variant, and classify its outcome. + + Returns the record together with one of ``annotated``, ``unannotated``, or ``errored``. Nothing raises: + a failure past the first record would end the body mid-stream, and since the 200 and its headers went + out with the first chunk the consumer has no way to be told and simply receives a short file. A failed + variant is reported in-band instead, as a record carrying an ``error`` object. This includes variants + failing serialization. + + A variant with no mapping data is *not* an error. It is an expected absence, reported as a null + annotation. + """ + variant_urn = mapped_variant.variant.urn + + try: + annotation = annotation_function(mapped_variant) + annotation_data = annotation.model_dump(exclude_none=True) if annotation else None + except MappingDataDoesntExistException: + logger.debug(f"Mapping data does not exist for variant {variant_urn}.") + return {"variant_urn": variant_urn, "annotation": None}, "unannotated" + except Exception as err: + logger.exception( + f"Failed to annotate variant {variant_urn}; streaming it as an error record.", + extra=logging_context(), + ) + return { + "variant_urn": variant_urn, + "annotation": None, + "error": {"type": type(err).__name__, "detail": str(err)}, + }, "errored" + + if annotation_data is None: + return {"variant_urn": variant_urn, "annotation": None}, "unannotated" + + return {"variant_urn": variant_urn, "annotation": annotation_data}, "annotated" + + def _stream_generated_annotations(mapped_variants, annotation_function): """ Generator function to stream annotations as pure NDJSON data. @@ -1277,35 +1317,23 @@ def _stream_generated_annotations(mapped_variants, annotation_function): - X-Processing-Started: ISO timestamp when processing began - X-Stream-Type: Type of annotation being streamed + Emits exactly one record per mapped variant, so a body holding fewer lines than ``X-Total-Count`` is + a truncated one. Outcome counts are logged rather than appended to the body, which keeps every line a + variant record and keeps this format identical to the public dump's ``va/{urn}.va.ndjson``. + Progress updates are sent as structured log events that can be consumed via Server-Sent Events if needed. """ start_time = time.time() total_variants = len(mapped_variants) processed_count = 0 + outcome_counts = {"annotated": 0, "unannotated": 0, "errored": 0} logger.info(f"Starting streaming processing of {total_variants} mapped variants") - for i, mv in enumerate(mapped_variants): - try: - annotation = annotation_function(mv) - except MappingDataDoesntExistException: - logger.debug(f"Mapping data does not exist for variant {mv.variant.urn}.") - annotation = None - except Exception: - # Raising here would end the body mid-stream. The 200 and its headers went out with the first - # chunk, so the client has no way to be told and simply receives a short file. Report the - # variant as unannotated and keep going, so one bad variant cannot truncate a whole download. - logger.exception( - f"Failed to annotate variant {mv.variant.urn}; streaming it as unannotated.", - extra=logging_context(), - ) - annotation = None + for mv in mapped_variants: + result, outcome = _annotation_stream_record(mv, annotation_function) + outcome_counts[outcome] += 1 - # Send pure result data (no wrapper) - result = { - "variant_urn": mv.variant.urn, - "annotation": annotation.model_dump(exclude_none=True) if annotation else None, - } yield json.dumps(result, default=str) + "\n" # Log server-side progress @@ -1332,6 +1360,7 @@ def _stream_generated_annotations(mapped_variants, annotation_function): { "stream_completion": { "total_processed": processed_count, + **outcome_counts, "total_time": round(total_time, 2), "average_time_per_variant": average_time_per_variant, "final_rate": final_rate, @@ -1340,7 +1369,8 @@ def _stream_generated_annotations(mapped_variants, annotation_function): } ) logger.info( - f"Completed streaming {processed_count} variants in {total_time:.2f} seconds (avg: {average_time_per_variant:.4f}s/variant)", + f"Completed streaming {processed_count} variants in {total_time:.2f} seconds " + f"({outcome_counts['errored']} errored, avg: {average_time_per_variant:.4f}s/variant)", extra=logging_context(), ) @@ -1374,8 +1404,8 @@ def get_score_set_annotated_variants( JSON (NDJSON) format for efficient processing of large datasets. NDJSON Response Format: - Each line in the response corresponds to a mapped variant and contains a JSON - object with the following structure: + Each line corresponds to a mapped variant and contains a JSON object with the following + structure: ``` { "variant_urn": "", @@ -1385,6 +1415,20 @@ def get_score_set_annotated_variants( } ``` + `annotation` is null where the variant has no mapping data to annotate, or no pathogenicity statements apply + to it. A variant whose annotation could not be built is reported in-band rather than by + truncating the stream, and carries an additional `error` object: + ``` + { + "variant_urn": "", + "annotation": null, + "error": {"type": "", "detail": ""} + } + ``` + + Every line is a variant record: a response holds exactly `X-Total-Count` lines, so a shorter + body is a truncated one. + Args: urn (str): The Uniform Resource Name (URN) of the score set to retrieve annotated variants for. @@ -1474,8 +1518,8 @@ def get_score_set_annotated_variants_functional_statement( JSON (NDJSON) format. NDJSON Response Format: - Each line in the response corresponds to a mapped variant and contains a JSON - object with the following structure: + Each line corresponds to a mapped variant and contains a JSON object with the following + structure: ``` { "variant_urn": "", @@ -1485,6 +1529,20 @@ def get_score_set_annotated_variants_functional_statement( } ``` + `annotation` is null where the variant has no mapping data to annotate, or no functional impact statements apply + to it. A variant whose annotation could not be built is reported in-band rather than by + truncating the stream, and carries an additional `error` object: + ``` + { + "variant_urn": "", + "annotation": null, + "error": {"type": "", "detail": ""} + } + ``` + + Every line is a variant record: a response holds exactly `X-Total-Count` lines, so a shorter + body is a truncated one. + Args: urn (str): The unique resource name (URN) identifying the score set. db (Session): Database session dependency for querying data. @@ -1568,8 +1626,8 @@ def get_score_set_annotated_variants_functional_study_result( (NDJSON) format for efficient streaming of large datasets. NDJSON Response Format: - Each line in the response corresponds to a mapped variant and contains a JSON - object with the following structure: + Each line corresponds to a mapped variant and contains a JSON object with the following + structure: ``` { "variant_urn": "", @@ -1579,6 +1637,20 @@ def get_score_set_annotated_variants_functional_study_result( } ``` + `annotation` is null where the variant has no mapping data to annotate, or no study results apply + to it. A variant whose annotation could not be built is reported in-band rather than by + truncating the stream, and carries an additional `error` object: + ``` + { + "variant_urn": "", + "annotation": null, + "error": {"type": "", "detail": ""} + } + ``` + + Every line is a variant record: a response holds exactly `X-Total-Count` lines, so a shorter + body is a truncated one. + Args: urn (str): The URN (Uniform Resource Name) of the score set to retrieve variants for. db (Session): Database session dependency for querying the database. diff --git a/tests/routers/test_score_set.py b/tests/routers/test_score_set.py index f58aa940c..26a45bb32 100644 --- a/tests/routers/test_score_set.py +++ b/tests/routers/test_score_set.py @@ -17,6 +17,8 @@ cdot = pytest.importorskip("cdot") fastapi = pytest.importorskip("fastapi") +from mavedb.lib.annotation.annotate import variant_study_result +from mavedb.lib.annotation.exceptions import MappingDataDoesntExistException from mavedb.lib.exceptions import NonexistentOrcidUserError from mavedb.lib.validation.urn_re import MAVEDB_EXPERIMENT_URN_RE, MAVEDB_SCORE_SET_URN_RE, MAVEDB_TMP_URN_RE from mavedb.models.enums.processing_state import ProcessingState @@ -28,6 +30,7 @@ from mavedb.models.mapped_variant import MappedVariant as MappedVariantDbModel from mavedb.models.score_set import ScoreSet as ScoreSetDbModel from mavedb.models.variant import Variant as VariantDbModel +from mavedb.routers.score_sets import _annotation_stream_record from mavedb.view_models.orcid import OrcidUser from mavedb.view_models.score_set import ScoreSet, ScoreSetCreate from tests.helpers.constants import ( @@ -64,7 +67,9 @@ VALID_CLINGEN_CA_ID, ) from tests.helpers.dependency_overrider import DependencyOverrider +from tests.helpers.mocks.factories import create_mock_mapped_variant from tests.helpers.util.common import ( + create_failing_side_effect, deepcamelize, parse_ndjson_response, update_expected_response_for_created_resources, @@ -4671,6 +4676,164 @@ def test_annotated_functional_study_result_exists_for_score_set_when_some_varian assert annotated_variant.get("type") == "ExperimentalVariantFunctionalImpactStudyResult" +def test_annotation_stream_reports_a_failing_variant_instead_of_truncating( + client, session, data_provider, data_files, setup_router_db +): + """One variant that cannot be annotated must not cost the consumer the rest of the download.""" + experiment = create_experiment(client) + score_set = create_seq_score_set_with_mapped_variants( + client, session, data_provider, experiment["urn"], data_files / "scores.csv" + ) + + failing_annotation = create_failing_side_effect( + # Representative of lib/annotation/util.py, which raises this on an unrecognized VRS Allele state. + ValueError("Unsupported VRS state type"), + variant_study_result, + fail_on_call=2, + ) + + with patch("mavedb.routers.score_sets.variant_study_result", failing_annotation): + response = client.get(f"/api/v1/score-sets/{score_set['urn']}/annotated-variants/study-result") + + assert response.status_code == 200 + + response_data = parse_ndjson_response(response) + assert len(response_data) == score_set["numVariants"] + + errored = [record for record in response_data if "error" in record] + assert len(errored) == 1 + assert errored[0]["annotation"] is None + assert errored[0]["error"] == {"type": "ValueError", "detail": "Unsupported VRS state type"} + + for record in response_data: + if "error" not in record: + assert record["annotation"].get("type") == "ExperimentalVariantFunctionalImpactStudyResult" + + +def test_annotation_stream_emits_one_record_per_variant_despite_a_failure( + client, session, data_provider, data_files, setup_router_db +): + """Every line is a variant record, so a body shorter than X-Total-Count is a truncated one.""" + experiment = create_experiment(client) + score_set = create_seq_score_set_with_mapped_variants( + client, session, data_provider, experiment["urn"], data_files / "scores.csv" + ) + unmapped_variant = clear_first_mapped_variant_post_mapped(session, score_set["urn"]) + assert unmapped_variant is not None + + failing_annotation = create_failing_side_effect( + ValueError("Unsupported VRS state type"), variant_study_result, fail_on_call=2 + ) + + with patch("mavedb.routers.score_sets.variant_study_result", failing_annotation): + response = client.get(f"/api/v1/score-sets/{score_set['urn']}/annotated-variants/study-result") + + total_count = int(response.headers["X-Total-Count"]) + assert total_count == score_set["numVariants"] + + response_data = parse_ndjson_response(response) + assert len(response_data) == total_count + assert all("variant_urn" in record for record in response_data) + + # A variant with no mapping data is an expected absence, distinguishable from the failure. + errored = [record for record in response_data if "error" in record] + unannotated = [record for record in response_data if record["annotation"] is None and "error" not in record] + assert len(errored) == 1 + assert len(unannotated) == 1 + assert unannotated[0]["variant_urn"] == unmapped_variant.urn + + +######################################################################################################################## +# Building individual annotation stream records +# +# Driven directly rather than over HTTP: these branches are about how a failure is classified, and +# reaching any one of them through the endpoint costs the whole app and a database. +######################################################################################################################## + + +class _StubAnnotation: + def model_dump(self, **kwargs): + return {"type": "Stub"} + + +class _UndumpableAnnotation: + def model_dump(self, **kwargs): + raise ValueError("Extension.value is required") + + +def _annotation_raising(exception): + """An annotation function that fails.""" + + def annotate(_mapped_variant): + raise exception + + return annotate + + +@pytest.fixture +def mock_mapped_variant_for_stream(): + return create_mock_mapped_variant(clingen_allele_id="CA123456") + + +def test_annotation_stream_record_serializes_a_successful_annotation(mock_mapped_variant_for_stream): + record, outcome = _annotation_stream_record(mock_mapped_variant_for_stream, lambda mv: _StubAnnotation()) + + assert outcome == "annotated" + assert record == {"variant_urn": mock_mapped_variant_for_stream.variant.urn, "annotation": {"type": "Stub"}} + + +def test_annotation_stream_record_treats_a_null_annotation_as_unannotated(mock_mapped_variant_for_stream): + """A variant the annotation layer declines to annotate is an expected outcome, not a failure.""" + record, outcome = _annotation_stream_record(mock_mapped_variant_for_stream, lambda mv: None) + + assert outcome == "unannotated" + assert record == {"variant_urn": mock_mapped_variant_for_stream.variant.urn, "annotation": None} + + +def test_annotation_stream_record_treats_missing_mapping_data_as_unannotated(mock_mapped_variant_for_stream): + # Preserved deliberately: a missing mapping is an expected absence, and reporting it as an error would + # tell consumers a variant failed when nothing went wrong. + record, outcome = _annotation_stream_record( + mock_mapped_variant_for_stream, _annotation_raising(MappingDataDoesntExistException("no post-mapped allele")) + ) + + assert outcome == "unannotated" + assert "error" not in record + assert record["annotation"] is None + + +@pytest.mark.parametrize( + "exception", + [ + # lib/annotation/study_result.py, on absent or malformed score data. + KeyError("score"), + TypeError("'NoneType' object is not subscriptable"), + # lib/annotation/util.py, on an unrecognized VRS Allele state type. + ValueError("Unsupported VRS state type"), + IndexError("list index out of range"), + ], +) +def test_annotation_stream_record_reports_any_other_failure_as_an_error(mock_mapped_variant_for_stream, exception): + record, outcome = _annotation_stream_record(mock_mapped_variant_for_stream, _annotation_raising(exception)) + + assert outcome == "errored" + assert record["variant_urn"] == mock_mapped_variant_for_stream.variant.urn + assert record["annotation"] is None + assert record["error"] == {"type": type(exception).__name__, "detail": str(exception)} + + +def test_annotation_stream_record_reports_a_serialization_failure_as_an_error(mock_mapped_variant_for_stream): + """An emitted object that no longer dumps is the shape-dependent failure this stream must survive. + + Commit 5c155f4d fixed exactly this: a required field combined with `exclude_none` produced an object + that built successfully and then failed on the way out. + """ + record, outcome = _annotation_stream_record(mock_mapped_variant_for_stream, lambda mv: _UndumpableAnnotation()) + + assert outcome == "errored" + assert record["error"] == {"type": "ValueError", "detail": "Extension.value is required"} + + ######################################################################################################################## # Fetching gnomad variants for a score set ######################################################################################################################## From 81536f7434cef8d20246e2273011bee6d8fce33d Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Mon, 10 Aug 2026 18:51:42 -0700 Subject: [PATCH 3/7] test(annotation): assert emitted annotations round-trip, across a list of variant shapes Annotation tests all ran against one variant shape, so a structural regression that only appears for a particular stored payload passed the whole suite. Two have reached production: a reference-identical variant whose VRS state is a ReferenceLengthExpression (84125081), and a required Extension.value combined with a null baseline score stripped by model_dump(exclude_none=True) (5c155f4d). Parametrize the four annotation entry points over 13 mapped-variant shapes and assert one contract per pair: an object that survives serialization and re-validation, or None. Never raises. Reverting either fix above now fails this suite. round_trip_annotation compares emitted JSON to re-emitted JSON rather than comparing objects. VA-Spec declares Statement.hasEvidenceLines as list[EvidenceLine], so a VariantPathogenicityEvidenceLine re-validates as its base class -- verified to lose no data and produce identical JSON, so object equality would report a non-defect on every pathogenicity statement. Shapes for gnomAD records, ClinVar controls, hgvs_g/hgvs_p, and a missing score key are deliberately absent; the first three are never read by the annotation layer, and dataframe validation rejects a score file with no score column. Reasoning is recorded in the module. --- src/mavedb/lib/annotation/conformance.py | 58 ++++++++ tests/helpers/constants.py | 42 ++++++ tests/helpers/mocks/factories.py | 30 +++- tests/helpers/variant_shapes.py | 176 +++++++++++++++++++++++ tests/lib/annotation/conftest.py | 12 ++ tests/lib/annotation/test_annotate.py | 14 +- tests/lib/annotation/test_conformance.py | 86 +++++++++++ 7 files changed, 398 insertions(+), 20 deletions(-) create mode 100644 src/mavedb/lib/annotation/conformance.py create mode 100644 tests/helpers/variant_shapes.py create mode 100644 tests/lib/annotation/test_conformance.py diff --git a/src/mavedb/lib/annotation/conformance.py b/src/mavedb/lib/annotation/conformance.py new file mode 100644 index 000000000..851980cb5 --- /dev/null +++ b/src/mavedb/lib/annotation/conformance.py @@ -0,0 +1,58 @@ +"""Structural conformance checks for emitted VA-Spec annotations. + +An annotation that constructs successfully is not necessarily one a consumer can read back. Two +defects of exactly that shape have reached production: a required ``Extension.value`` combined with a +null baseline score stripped by ``model_dump(exclude_none=True)`` (fixed in 5c155f4d), and a +reference-identical variant whose VRS state is a ``ReferenceLengthExpression`` (fixed in 84125081). +Neither was caught by a test that only asserted the object had been built. + +This module is declared here so it can be run from both the test suite and from script based conformance +checks. +""" + +import json +from typing import TypeVar + +from ga4gh.va_spec.acmg_2015 import VariantPathogenicityStatement +from ga4gh.va_spec.base.core import ExperimentalVariantFunctionalImpactStudyResult, Statement + +Annotation = TypeVar( + "Annotation", ExperimentalVariantFunctionalImpactStudyResult, Statement, VariantPathogenicityStatement +) + + +class AnnotationRoundTripError(Exception): + """An emitted annotation did not survive serialization and re-validation.""" + + +def round_trip_annotation(annotation: Annotation) -> Annotation: + """Serialize an annotation the way the API emits it, then read it back. + + Mirrors the emission path exactly — ``model_dump(exclude_none=True)`` then ``json.dumps(default=str)`` + — because the failures worth catching are the ones those two steps introduce. + + The assertion is that emission is a fixed point: the re-validated object dumps to the same JSON it + was parsed from. Object equality would be the wrong test. VA-Spec declares container fields in terms + of base classes — ``Statement.hasEvidenceLines`` is a ``list[EvidenceLine]`` — so a + ``VariantPathogenicityEvidenceLine`` re-validates as a plain ``EvidenceLine``. That narrowing loses + no data and produces byte-identical JSON, so it is not a defect this check should report. + + Returns the re-validated object. + + Raises: + AnnotationRoundTripError: The emitted JSON did not re-validate, or re-validated to something + that serializes differently than what was emitted. + """ + model = type(annotation) + emitted = json.dumps(annotation.model_dump(exclude_none=True), default=str) + + try: + reparsed = model.model_validate(json.loads(emitted)) + except Exception as err: + raise AnnotationRoundTripError(f"{model.__name__} did not re-validate after emission: {err}") from err + + re_emitted = json.dumps(reparsed.model_dump(exclude_none=True), default=str) + if re_emitted != emitted: + raise AnnotationRoundTripError(f"{model.__name__} did not survive a round trip unchanged.") + + return reparsed diff --git a/tests/helpers/constants.py b/tests/helpers/constants.py index d582312b9..faabe6d18 100644 --- a/tests/helpers/constants.py +++ b/tests/helpers/constants.py @@ -167,6 +167,48 @@ }, } +# A genomic mapping: an hgvs.g expression against a chromosome accession, rather than the protein-level +# hgvs.p the other post-mapped constants carry. +TEST_VALID_POST_MAPPED_VRS_ALLELE_GENOMIC = { + "id": TEST_GA4GH_IDENTIFIER, + "type": "Allele", + "state": {"type": "LiteralSequenceExpression", "sequence": "G"}, + "digest": TEST_GA4GH_DIGEST, + "location": { + "id": TEST_SEQUENCE_LOCATION_ACCESSION, + "end": 23536836, + "type": "SequenceLocation", + "start": 23536835, + "digest": TEST_GA4GH_DIGEST, + "sequenceReference": { + "type": "SequenceReference", + "label": "NC_000018.10", + "refgetAccession": TEST_REFGET_ACCESSION, + }, + }, + "expressions": [{"value": "NC_000018.10:g.23536836C>G", "syntax": "hgvs.g"}], +} + +# The minimum a mapper can store: an allele with no expressions and no reference-sequence extension. +# Annotation must not assume either is present. +TEST_VALID_POST_MAPPED_VRS_ALLELE_DIGEST_ONLY = { + "id": TEST_GA4GH_IDENTIFIER, + "type": "Allele", + "state": {"type": "LiteralSequenceExpression", "sequence": "F"}, + "digest": TEST_GA4GH_DIGEST, + "location": { + "id": TEST_SEQUENCE_LOCATION_ACCESSION, + "end": 6, + "type": "SequenceLocation", + "start": 5, + "digest": TEST_GA4GH_DIGEST, + "sequenceReference": { + "type": "SequenceReference", + "refgetAccession": TEST_REFGET_ACCESSION, + }, + }, +} + TEST_PUBMED_PUBLICATION = { "identifier": TEST_PUBMED_IDENTIFIER, "db_name": "PubMed", diff --git a/tests/helpers/mocks/factories.py b/tests/helpers/mocks/factories.py index 76b011940..4fbc5bfa3 100644 --- a/tests/helpers/mocks/factories.py +++ b/tests/helpers/mocks/factories.py @@ -21,6 +21,9 @@ create_sealed_mock, ) +# Sentinel for optional overrides whose meaningful values include None. +_UNSET = object() + # --------------------------------------------------------------------------- # License and Legal Helpers # --------------------------------------------------------------------------- @@ -290,11 +293,15 @@ def create_mock_score_calibration_with_ranges(score_set=None, user=None): # --------------------------------------------------------------------------- -def create_mock_variant(urn="test:variant", score=0.5, score_set=None): - """Create a mock Variant with specified properties.""" +def create_mock_variant(urn="test:variant", score=0.5, score_set=None, data=_UNSET): + """Create a mock Variant with specified properties. + + ``data`` defaults to a well-formed ``score_data`` built from ``score``. Pass it explicitly to model a + variant whose score data is absent or malformed, which ``score`` cannot express. + """ return create_sealed_mock( urn=urn, - data={"score_data": {"score": score}}, + data={"score_data": {"score": score}} if data is _UNSET else data, score_set=score_set or create_mock_score_set(), id=1, score=score, @@ -315,17 +322,26 @@ def create_mock_mapped_variant( mapped_date=None, clingen_allele_id=None, score_set=None, + pre_mapped=None, + post_mapped=None, + variant_data=_UNSET, ): - """Create a mock MappedVariant with specified properties.""" - mock_variant = create_mock_variant(urn=urn, score=score, score_set=score_set) + """Create a mock MappedVariant with specified properties. + + ``pre_mapped`` and ``post_mapped`` default to the VRS 2.x constants; pass a payload to build a + variant of a different shape. ``variant_data`` overrides the variant's ``data`` wholesale, which is + how a variant with an absent or non-numeric score is expressed — ``score`` alone can only produce a + well-formed ``score_data``. + """ + mock_variant = create_mock_variant(urn=urn, score=score, score_set=score_set, data=variant_data) return create_sealed_mock( variant=mock_variant, mapping_api_version=mapping_api_version, mapped_date=mapped_date or datetime(2024, 1, 15, 10, 30, 0), clingen_allele_id=clingen_allele_id, - pre_mapped=TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS2_X, - post_mapped=TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X, + pre_mapped=TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS2_X if pre_mapped is None else pre_mapped, + post_mapped=TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X if post_mapped is None else post_mapped, ) diff --git a/tests/helpers/variant_shapes.py b/tests/helpers/variant_shapes.py new file mode 100644 index 000000000..b9a8ef790 --- /dev/null +++ b/tests/helpers/variant_shapes.py @@ -0,0 +1,176 @@ +"""The mapped-variant shapes every annotation surface has to survive. + +Annotation tests otherwise all run against one variant, so a defect that only appears for a particular +stored payload passes the whole suite. Both production failures this list exists to catch were of that +kind: a reference-identical variant whose VRS state is a ``ReferenceLengthExpression`` (84125081) and a +null baseline score stripped on the way out (5c155f4d). + +Lives here rather than under ``tests/lib/annotation`` because the CSV surfaces will consume the same +list, and a shape list inside the annotation package would have to move when they do. + +Adding a shape is one entry in ``VARIANT_SHAPES``. It applies to every annotation surface at once. + +Expected to be short-lived. This builds on the non-DB mock factories in ``tests/helpers/mocks/factories.py``, +hand-rolling the override plumbing — ``kwargs`` forwarded to a factory, plus a ``mutate`` hook for the axes +a factory signature cannot reach. #782 is an open decision on how the suite should construct test objects at +all (``factory_boy`` versus explicit scenario builders). If ``factory_boy`` wins, ``VariantShape.kwargs`` +becomes factory params or traits and ``mutate`` becomes a post-generation hook, and this module reduces +to the list itself. The parametrize-over-a-list structure is worth keeping either way; the plumbing here +is not. + +Note that #782 is chiefly about *DB-backed* construction and explicitly keeps that layer distinct from +these mock factories, so adoption there does not automatically retire this — but the two should be +reconciled rather than left to drift. +""" + +from dataclasses import dataclass, field +from typing import Any, Callable, Optional + +from tests.helpers.constants import ( + TEST_VALID_POST_MAPPED_VRS_ALLELE, + TEST_VALID_POST_MAPPED_VRS_ALLELE_DIGEST_ONLY, + TEST_VALID_POST_MAPPED_VRS_ALLELE_GENOMIC, + TEST_VALID_POST_MAPPED_VRS_ALLELE_LENGTH_EXPRESSION, + TEST_VALID_POST_MAPPED_VRS_ALLELE_RLE, + TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS1_X, + TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X, + TEST_VALID_POST_MAPPED_VRS_CIS_PHASED_BLOCK, + TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS1_X, + TEST_VALID_PRE_MAPPED_VRS_CIS_PHASED_BLOCK, +) + +# Every shape sets this unless it is the axis under test, so that a variant carrying no ClinGen id is a +# deliberate case rather than an accident of the factory default. +DEFAULT_CLINGEN_ALLELE_ID = "CA123456" + + +@dataclass(frozen=True) +class VariantShape: + """One mapped-variant configuration, and how to build it from a mock factory.""" + + name: str + why: str + kwargs: dict[str, Any] = field(default_factory=dict) + #: Applied after construction, for axes the factory signature does not reach. + mutate: Optional[Callable[[Any], None]] = None + + def build(self, factory: Callable[..., Any]): + """Build this shape with *factory*, one of the ``create_mock_mapped_variant*`` functions.""" + mapped_variant = factory(**{"clingen_allele_id": DEFAULT_CLINGEN_ALLELE_ID, **self.kwargs}) + if self.mutate is not None: + self.mutate(mapped_variant) + return mapped_variant + + +def _only_target_gene(mapped_variant): + return mapped_variant.variant.score_set.target_genes[0] + + +def _drop_gene_symbol(mapped_variant) -> None: + """Force the fallback from the mapped HGNC name down to the target's own name. + + ``post_mapped_metadata`` has to be set explicitly: the mock factory leaves it unset, and an unset + attribute on a MagicMock is a truthy mock rather than the empty metadata a real target would have. + """ + target = _only_target_gene(mapped_variant) + target.mapped_hgnc_name = None + target.post_mapped_metadata = {} + + +def _drop_baseline_score(mapped_variant) -> None: + """A calibration with no baseline score. + + ``Extension.value`` is required, so an extension built around a null baseline score was stripped by + ``model_dump(exclude_none=True)`` and the emitted object then refused to re-parse. That is the defect + 5c155f4d fixed, and this is the shape that reaches it. + """ + for calibration in mapped_variant.variant.score_set.score_calibrations: + calibration.baseline_score = None + calibration.baseline_score_description = None + + +def _make_non_coding(mapped_variant) -> None: + target = _only_target_gene(mapped_variant) + target.category = "Regulatory" + target.mapped_hgnc_name = None + target.post_mapped_metadata = {"genomic": {"sequence_id": "ga4gh:SQ.test"}} + + +VARIANT_SHAPES: list[VariantShape] = [ + VariantShape( + name="vrs2_allele", + why="the current default: a VRS 2.x allele with a literal sequence state", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X}, + ), + VariantShape( + name="vrs1_allele", + why="VRS 1.x nests the allele under a `variation` key", + kwargs={ + "pre_mapped": TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS1_X, + "post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS1_X, + }, + ), + VariantShape( + name="protein_allele", + why="an hgvs.p expression carrying neither `type` nor `syntax_version`", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE}, + ), + VariantShape( + name="genomic_expression", + why="an hgvs.g expression against a chromosome accession rather than a protein one", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_GENOMIC}, + ), + VariantShape( + name="reference_length_expression", + why="reference-identical variants store an RLE state; 84125081 fixed a 500 on exactly this", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_RLE}, + ), + VariantShape( + name="length_expression", + why="a LengthExpression state, the third of the three states util.py accepts", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_LENGTH_EXPRESSION}, + ), + VariantShape( + name="cis_phased_block", + why="a haplotype resolves to a CisPhasedBlock rather than a bare Allele", + kwargs={ + "pre_mapped": TEST_VALID_PRE_MAPPED_VRS_CIS_PHASED_BLOCK, + "post_mapped": TEST_VALID_POST_MAPPED_VRS_CIS_PHASED_BLOCK, + }, + ), + VariantShape( + name="digest_only_post_mapped", + why="an allele with no expressions and no reference-sequence extension", + kwargs={"post_mapped": TEST_VALID_POST_MAPPED_VRS_ALLELE_DIGEST_ONLY}, + ), + VariantShape( + name="null_score", + why="an NA score, which `exclude_none=True` strips on the way out", + kwargs={"score": None}, + ), + VariantShape( + name="absent_baseline_score", + why="a calibration with no baseline score; 5c155f4d fixed output that no longer re-parsed", + mutate=_drop_baseline_score, + ), + VariantShape( + name="absent_clingen_allele_id", + why="no ClinGen allele id, so the variant has no canonical IRI to report", + kwargs={"clingen_allele_id": None}, + ), + VariantShape( + name="absent_gene_symbol", + why="no mapped HGNC name, falling back to the target's own name", + mutate=_drop_gene_symbol, + ), + VariantShape( + name="non_coding_target", + why="a regulatory target, whose identifier comes from post-mapped metadata", + mutate=_make_non_coding, + ), +] + + +def shape_ids() -> list[str]: + """Shape names, for use as pytest parametrize ids.""" + return [shape.name for shape in VARIANT_SHAPES] diff --git a/tests/lib/annotation/conftest.py b/tests/lib/annotation/conftest.py index 29a056c67..d93e66c4b 100644 --- a/tests/lib/annotation/conftest.py +++ b/tests/lib/annotation/conftest.py @@ -9,6 +9,7 @@ import pytest +from mavedb.lib.annotation.util import CALIBRATION_SCOPE_EXTENSION_NAME from tests.helpers.constants import PRIVATE_CALIBRATION_OWNER_ID from tests.helpers.mocks.factories import ( create_mock_mapped_variant, @@ -23,6 +24,17 @@ pass +def scope_of(annotation) -> str: + """The disclosed principal of an annotation, which every emitted object must carry.""" + scopes = [ + extension.value + for extension in (annotation.extensions or []) + if extension.name == CALIBRATION_SCOPE_EXTENSION_NAME + ] + assert len(scopes) == 1, f"expected exactly one calibration scope extension, found {scopes}" + return scopes[0] + + def make_private(mapped_variant, *, owner_id: int = PRIVATE_CALIBRATION_OWNER_ID): """Mark every calibration on a mapped variant's score set private, owned by ``owner_id``. diff --git a/tests/lib/annotation/test_annotate.py b/tests/lib/annotation/test_annotate.py index 0d0b091f5..cbfc19f2a 100644 --- a/tests/lib/annotation/test_annotate.py +++ b/tests/lib/annotation/test_annotate.py @@ -20,19 +20,7 @@ variant_pathogenicity_statement, variant_study_result, ) -from mavedb.lib.annotation.util import CALIBRATION_SCOPE_EXTENSION_NAME -from tests.lib.annotation.conftest import admin_principal, make_private, owner_principal - - -def scope_of(annotation) -> str: - """The disclosed principal of an annotation, which every emitted object must carry.""" - scopes = [ - extension.value - for extension in (annotation.extensions or []) - if extension.name == CALIBRATION_SCOPE_EXTENSION_NAME - ] - assert len(scopes) == 1, f"expected exactly one calibration scope extension, found {scopes}" - return scopes[0] +from tests.lib.annotation.conftest import admin_principal, make_private, owner_principal, scope_of @pytest.mark.unit diff --git a/tests/lib/annotation/test_conformance.py b/tests/lib/annotation/test_conformance.py new file mode 100644 index 000000000..ed83878ff --- /dev/null +++ b/tests/lib/annotation/test_conformance.py @@ -0,0 +1,86 @@ +"""Structural conformance of emitted annotations, across every mapped-variant shape. + +The contract asserted here is deliberately narrow and applies to every (surface, shape) pair: an +annotation function returns either an object that survives serialization and re-validation, or None. +It never raises. Content is asserted elsewhere; this file exists because content assertions on a +single variant shape cannot catch a payload-dependent structural regression. +""" + +# ruff: noqa: E402 + +import pytest + +pytest.importorskip("psycopg2") +pytest.importorskip("fastapi") + +from mavedb.lib.annotation.annotate import ( + variant_functional_impact_statement, + variant_highest_level_annotation, + variant_pathogenicity_statement, + variant_study_result, +) +from mavedb.lib.annotation.conformance import round_trip_annotation +from tests.helpers.mocks.factories import ( + create_mock_mapped_variant, + create_mock_mapped_variant_with_functional_calibration_score_set, + create_mock_mapped_variant_with_pathogenicity_calibration_score_set, +) +from tests.helpers.variant_shapes import VARIANT_SHAPES, shape_ids +from tests.lib.annotation.conftest import scope_of + +# Each surface is paired with the factory that can actually exercise it: a statement built on a score +# set with no calibrations returns None for every shape, which would test nothing. +ANNOTATION_SURFACES = [ + ("study_result", variant_study_result, create_mock_mapped_variant), + ( + "functional_impact_statement", + variant_functional_impact_statement, + create_mock_mapped_variant_with_functional_calibration_score_set, + ), + ( + "pathogenicity_statement", + variant_pathogenicity_statement, + create_mock_mapped_variant_with_pathogenicity_calibration_score_set, + ), + ( + "highest_level_annotation", + variant_highest_level_annotation, + create_mock_mapped_variant_with_pathogenicity_calibration_score_set, + ), +] + +SURFACE_IDS = [name for name, _, _ in ANNOTATION_SURFACES] + + +@pytest.mark.unit +@pytest.mark.parametrize("shape", VARIANT_SHAPES, ids=shape_ids()) +@pytest.mark.parametrize("surface", ANNOTATION_SURFACES, ids=SURFACE_IDS) +class TestAnnotationConformance: + def test_annotation_round_trips_or_is_none(self, surface, shape): + """Never raises, and anything emitted can be read back.""" + _, annotate, factory = surface + + annotation = annotate(shape.build(factory)) + + if annotation is None: + pytest.skip(f"{shape.name} produces no annotation on this surface") + round_trip_annotation(annotation) + + def test_calibration_scope_survives_the_round_trip(self, surface, shape): + """`mavedb_calibration_scope` is emitted unconditionally, so a missing scope is never ambiguous + between "public" and "produced before disclosure existed". It has to come back too.""" + _, annotate, factory = surface + + annotation = annotate(shape.build(factory)) + + if annotation is None: + pytest.skip(f"{shape.name} produces no annotation on this surface") + assert scope_of(round_trip_annotation(annotation)) == scope_of(annotation) + + +@pytest.mark.unit +def test_every_surface_emits_something_for_at_least_one_shape(): + """Guards the suite above: a surface that returned None everywhere would silently skip every case.""" + for name, annotate, factory in ANNOTATION_SURFACES: + emitted = [shape.name for shape in VARIANT_SHAPES if annotate(shape.build(factory)) is not None] + assert emitted, f"{name} emitted nothing for any shape; its conformance cases are all skips" From 51bb548362eeee875e581013c437b27f34489984 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Mon, 10 Aug 2026 21:09:45 -0700 Subject: [PATCH 4/7] feat(scripts): sweep the annotation surfaces across the published corpus The annotation test suite runs against constructed variant shapes. Real data holds shapes nobody thought to construct, and both VA-Spec serialization defects that reached production were of that kind. This walks the corpus and attempts every annotation surface, so those shapes get a chance to fail somewhere other than a user's download. One CSV row per attempted (score set, surface) pair, successes included: a report showing only failures cannot distinguish "nothing broke" from "nothing ran". Outcomes are ok, exception, schema_violation, or skipped, and the exit code is non-zero only for the two that mean something is broken -- most published score sets have no current mapped variants, so failing on skipped would make the exit code useless. The schema check reuses the round-trip helper the conformance tests use. No per-score-set sampling. An earlier draft sampled, but a sampled sweep can only report a score set as unbroken-where-sampled, and measurement put a full run at 43 minutes for 3.4M variants across three surfaces -- the right order for a pre-release check. --max-score-sets bounds how many score sets are looked at; every one it reports on is swept completely. Which exceptions mean "nothing to annotate" rather than "failed to annotate" now lives in EXPECTED_ABSENCE_EXCEPTIONS, since the streaming endpoints and this script both have to draw that line and a sweep that treated an expected absence as a failure would bury real defects in noise. --- src/mavedb/lib/annotation/exceptions.py | 8 + src/mavedb/routers/score_sets.py | 11 +- src/mavedb/scripts/export_sweep.py | 299 ++++++++++++++++++++++++ 3 files changed, 313 insertions(+), 5 deletions(-) create mode 100644 src/mavedb/scripts/export_sweep.py diff --git a/src/mavedb/lib/annotation/exceptions.py b/src/mavedb/lib/annotation/exceptions.py index 27511a56a..23662f713 100644 --- a/src/mavedb/lib/annotation/exceptions.py +++ b/src/mavedb/lib/annotation/exceptions.py @@ -1,2 +1,10 @@ class MappingDataDoesntExistException(ValueError): pass + + +#: Exceptions meaning a variant has nothing to annotate, as opposed to something going wrong while +#: annotating it. Every caller that reports failures has to tell the two apart: the streaming endpoints +#: emit an expected absence as a null annotation rather than an error record, and the corpus sweep counts +#: it as not-applicable rather than a defect. Adding an entry here changes both, and they must agree — +#: a sweep that treated an expected absence as a failure would bury real defects in noise. +EXPECTED_ABSENCE_EXCEPTIONS = (MappingDataDoesntExistException,) diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index d74a0abba..843db6f1b 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -28,7 +28,7 @@ variant_pathogenicity_statement, variant_study_result, ) -from mavedb.lib.annotation.exceptions import MappingDataDoesntExistException +from mavedb.lib.annotation.exceptions import EXPECTED_ABSENCE_EXCEPTIONS from mavedb.lib.authorization import ( get_current_user, get_principal, @@ -1280,16 +1280,17 @@ def _annotation_stream_record( variant is reported in-band instead, as a record carrying an ``error`` object. This includes variants failing serialization. - A variant with no mapping data is *not* an error. It is an expected absence, reported as a null - annotation. + A variant with nothing to annotate is *not* an error. It is an expected absence, reported as a null + annotation. Which exceptions mean that is defined once, in ``EXPECTED_ABSENCE_EXCEPTIONS``, because + the corpus sweep has to draw the same line and the two must not drift apart. """ variant_urn = mapped_variant.variant.urn try: annotation = annotation_function(mapped_variant) annotation_data = annotation.model_dump(exclude_none=True) if annotation else None - except MappingDataDoesntExistException: - logger.debug(f"Mapping data does not exist for variant {variant_urn}.") + except EXPECTED_ABSENCE_EXCEPTIONS: + logger.debug(f"Nothing to annotate for variant {variant_urn}.") return {"variant_urn": variant_urn, "annotation": None}, "unannotated" except Exception as err: logger.exception( diff --git a/src/mavedb/scripts/export_sweep.py b/src/mavedb/scripts/export_sweep.py new file mode 100644 index 000000000..aa96260ee --- /dev/null +++ b/src/mavedb/scripts/export_sweep.py @@ -0,0 +1,299 @@ +""" +Script that sweeps the VA-Spec annotation surfaces across the published corpus and reports what broke. + +Usage: +``` +python3 -m mavedb.scripts.export_sweep --output sweep.csv +``` + +The annotation test suite runs against constructed variant shapes. Real data contains shapes nobody +thought to construct, and both VA-Spec serialization defects that reached production were of that kind. +This walks the actual corpus and attempts every annotation surface, so the shapes that only exist in the +database get a chance to fail somewhere other than a user's download. + +Writes one CSV row per attempted (score set, surface) pair, successes included: a report showing only +failures cannot distinguish "nothing broke" from "nothing ran". Exits non-zero if any pair failed. + +Every current mapped variant of every published score set is attempted. There is deliberately no +per-score-set sampling: a sampled sweep can only ever report a score set as unbroken-where-sampled, +and the measured cost of doing all of it is well under an hour, which is the right order for a +pre-release or periodic check. + +The sweep is read-only and runs as an anonymous principal, matching what a public consumer receives. +Private score calibrations are therefore not exercised; a surface reachable only through a privileged +viewer will report as not-applicable here rather than being annotated. +""" + +import csv +import logging +import sys +from collections import Counter +from dataclasses import dataclass +from datetime import datetime, timezone +from functools import partial +from typing import Any, Callable, Optional + +import asyncclick as click +from sqlalchemy import select +from sqlalchemy.orm import Session + +from mavedb.lib.annotation.annotate import ( + variant_functional_impact_statement, + variant_pathogenicity_statement, + variant_study_result, +) +from mavedb.lib.annotation.conformance import AnnotationRoundTripError, round_trip_annotation +from mavedb.lib.annotation.exceptions import EXPECTED_ABSENCE_EXCEPTIONS +from mavedb.lib.permissions.principal import Principal +from mavedb.lib.score_sets import get_current_mapped_variants_for_annotation +from mavedb.models.mapped_variant import MappedVariant +from mavedb.models.score_set import ScoreSet +from mavedb.scripts.environment import script_environment, with_database_session + +logger = logging.getLogger(__name__) + +OK = "ok" +EXCEPTION = "exception" +SCHEMA_VIOLATION = "schema_violation" +SKIPPED = "skipped" + +#: Outcomes that mean a surface is broken, as opposed to merely inapplicable. These set the exit code. +FAILURE_OUTCOMES = (EXCEPTION, SCHEMA_VIOLATION) + +CSV_COLUMNS = [ + "score_set_urn", + "surface", + "outcome", + "variants_attempted", + "variants_annotated", + "variants_not_applicable", + "variants_failed", + "first_failing_variant_urn", + "exception_class", + "message", +] + + +def _surfaces(principal: Principal) -> list[tuple[str, Callable[[MappedVariant], Optional[Any]]]]: + """The three annotation surfaces the API streams, named as their endpoint path segments.""" + return [ + ("study-result", variant_study_result), + ("functional-statement", partial(variant_functional_impact_statement, principal=principal)), + ("pathogenicity-statement", partial(variant_pathogenicity_statement, principal=principal)), + ] + + +@dataclass +class SurfaceResult: + """What one annotation surface did across every current mapped variant of one score set.""" + + score_set_urn: str + surface: str + variants_attempted: int = 0 + variants_annotated: int = 0 + variants_not_applicable: int = 0 + variants_failed: int = 0 + outcome: str = OK + first_failing_variant_urn: str = "" + exception_class: str = "" + message: str = "" + #: Set when the pair was never attempted, e.g. the score set has no current mapped variants. + skip_reason: str = "" + + def record_failure(self, variant_urn: str, outcome: str, err: BaseException) -> None: + """Count a failure, keeping the first one's detail. + + The first is kept rather than the last because a surface that fails on one shape usually fails on + every variant of that shape, and a hundred identical messages are less useful than one plus a count. + """ + self.variants_failed += 1 + if self.outcome in FAILURE_OUTCOMES: + return + self.outcome = outcome + self.first_failing_variant_urn = variant_urn or "" + self.exception_class = type(err).__name__ + # Newlines would break the row apart for anyone reading the CSV with line-oriented tools. + self.message = " ".join(str(err).split()) + + def as_row(self) -> dict[str, Any]: + return { + "score_set_urn": self.score_set_urn, + "surface": self.surface, + "outcome": self.outcome, + "variants_attempted": self.variants_attempted, + "variants_annotated": self.variants_annotated, + "variants_not_applicable": self.variants_not_applicable, + "variants_failed": self.variants_failed, + "first_failing_variant_urn": self.first_failing_variant_urn, + "exception_class": self.exception_class, + "message": self.message or self.skip_reason, + } + + +@dataclass +class SweepTotals: + """Corpus-level counts, so a bounded run cannot be mistaken for a complete one.""" + + score_sets_published: int = 0 + score_sets_attempted: int = 0 + score_sets_skipped: int = 0 + variants_attempted: int = 0 + + +def sweep_surface( + score_set_urn: str, + surface: str, + annotate: Callable[[MappedVariant], Optional[Any]], + mapped_variants: list[MappedVariant], +) -> SurfaceResult: + """Attempt one annotation surface across every current mapped variant of one score set. + + Nothing raises out of here. A sweep that aborted on the first bad variant would report the corpus as + far healthier than it is. + """ + result = SurfaceResult( + score_set_urn=score_set_urn, + surface=surface, + variants_attempted=len(mapped_variants), + ) + + for mapped_variant in mapped_variants: + variant_urn = getattr(mapped_variant.variant, "urn", "") or "" + + try: + annotation = annotate(mapped_variant) + except EXPECTED_ABSENCE_EXCEPTIONS: + # An expected absence, drawn from the same definition the streaming endpoints use so the two + # cannot disagree. Counting it as a failure would bury real defects under millions of + # variants that simply have nothing to annotate. + result.variants_not_applicable += 1 + continue + except Exception as err: + result.record_failure(variant_urn, EXCEPTION, err) + continue + + if annotation is None: + # No calibration reaches this viewer, so this surface does not apply to this variant. + result.variants_not_applicable += 1 + continue + + try: + round_trip_annotation(annotation) + except AnnotationRoundTripError as err: + result.record_failure(variant_urn, SCHEMA_VIOLATION, err) + continue + except Exception as err: + # The conformance check itself blew up, which is still a defect in the emitted object. + result.record_failure(variant_urn, SCHEMA_VIOLATION, err) + continue + + result.variants_annotated += 1 + + return result + + +def published_score_sets(db: Session, max_score_sets: Optional[int]) -> list[ScoreSet]: + query = select(ScoreSet).where(ScoreSet.published_date.is_not(None)).order_by(ScoreSet.urn) + if max_score_sets is not None: + query = query.limit(max_score_sets) + return list(db.scalars(query).all()) + + +@script_environment.command() +@click.option( + "--max-score-sets", + default=None, + type=int, + help="Stop after this many published score sets. The only bound available, and deliberately so: " + "every score set the sweep reports on is swept completely, so a clean row means clean rather than " + "clean-where-sampled. Use it to smoke-test the sweep itself.", +) +@click.option( + "--output", + default=None, + help="CSV path. Defaults to export-sweep.YYYYMMDDHHMMSS.csv in the working directory.", +) +@with_database_session +def export_sweep(db: Session, max_score_sets: Optional[int], output: Optional[str]): + output_path = output or f"export-sweep.{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}.csv" + + # Matches the public data export: publishing a score set does not publish its calibrations. + principal = Principal() + surfaces = _surfaces(principal) + + score_sets = published_score_sets(db, max_score_sets) + totals = SweepTotals(score_sets_published=len(score_sets)) + + logger.info( + f"Sweeping {len(score_sets)} published score sets across {len(surfaces)} surfaces, " + "attempting every current mapped variant of each." + ) + if max_score_sets is not None: + logger.warning(f"Bounded to the first {max_score_sets} score sets by --max-score-sets; not full coverage.") + + rows: list[dict[str, Any]] = [] + + for index, score_set in enumerate(score_sets): + urn = score_set.urn or f"" + mapped_variants = list(get_current_mapped_variants_for_annotation(db, score_set)) + + if not mapped_variants: + totals.score_sets_skipped += 1 + for surface, _ in surfaces: + skipped = SurfaceResult( + score_set_urn=urn, + surface=surface, + outcome=SKIPPED, + skip_reason="no current mapped variants", + ) + rows.append(skipped.as_row()) + continue + + totals.score_sets_attempted += 1 + totals.variants_attempted += len(mapped_variants) + + for surface, annotate in surfaces: + result = sweep_surface(urn, surface, annotate, mapped_variants) + rows.append(result.as_row()) + + if result.outcome in FAILURE_OUTCOMES: + logger.error( + f"{urn} / {surface}: {result.outcome} on {result.variants_failed} of " + f"{result.variants_attempted} variant(s); first was " + f"{result.first_failing_variant_urn} ({result.exception_class}: {result.message})" + ) + + # The corpus is large enough that a silent run looks like a hung one. + if (index + 1) % 100 == 0: + logger.info(f"[{index + 1}/{len(score_sets)}] swept") + + # Each score set's variants are only needed while its surfaces are attempted. Without this the + # session accumulates the whole corpus. + db.expunge_all() + + with open(output_path, "w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=CSV_COLUMNS) + writer.writeheader() + writer.writerows(rows) + + # Counted off the emitted rows so the summary cannot disagree with the CSV it describes. + outcomes = Counter(row["outcome"] for row in rows) + failures = sum(outcomes[outcome] for outcome in FAILURE_OUTCOMES) + + logger.info(f"Wrote {len(rows)} rows to {output_path}") + logger.info( + f"Score sets: {totals.score_sets_published} published, {totals.score_sets_attempted} attempted, " + f"{totals.score_sets_skipped} skipped with no current mapped variants." + ) + logger.info(f"Variants: {totals.variants_attempted} attempted per surface, every one held by those score sets.") + logger.info("Outcomes: " + ", ".join(f"{outcome}={count}" for outcome, count in sorted(outcomes.items()))) + + if failures: + logger.error(f"{failures} (score set, surface) pair(s) failed. See {output_path}.") + sys.exit(1) + + logger.info("Every attempted surface annotated and round-tripped.") + + +if __name__ == "__main__": + export_sweep() From d955e68acc02697258eeff3c0e672e86f34569d4 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Mon, 10 Aug 2026 23:09:44 -0700 Subject: [PATCH 5/7] fix(csv): stop an unparseable post-mapped payload from aborting a whole export Running the variant shape list through the CSV row composer found that two post-mapped payload shapes raise out of a cell resolver: a VRS 1.x object, which hgvs_from_vrs_allele refuses deliberately, and an allele carrying no expressions key. The resolvers reach that parse whenever the stored hgvs_g or hgvs_p column is null, which is 3.9M and 3.0M of 4.2M current mapped variants, so the path is live even though neither payload shape occurs in the corpus today. Raised from inside one cell, either exception aborts the entire file -- every other variant in the score set's CSV, and the whole archive build for the public dump. _safe_hgvs_from_post_mapped absorbs both and returns None, which keeps the refusal without spending the caller's download on one variant's shape. Scoped to the CSV resolvers: the worker's mapping and ClinGen jobs call the same helper, and there an unparseable payload should surface rather than blank a cell. Also extends the shape list to the CSV surface. The mock mapped variant is annotation-shaped and left hgvs_g, hgvs_p, hgvs_c, hgvs_assay_level, and vep_functional_consequence unset -- and an unset attribute on a MagicMock is truthy, so the resolvers took the stored-column branch every time and the VRS fallback never ran. hgvs_g and hgvs_p now default to None deliberately, which is what makes the payload matter. --- src/mavedb/lib/csv/specs.py | 22 +++++++- tests/helpers/mocks/factories.py | 22 ++++++++ tests/helpers/variant_shapes.py | 33 +++++++++--- tests/lib/csv/test_columns.py | 91 ++++++++++++++++++++++++++++++++ 4 files changed, 158 insertions(+), 10 deletions(-) diff --git a/src/mavedb/lib/csv/specs.py b/src/mavedb/lib/csv/specs.py index 88e6d5f12..b309c674b 100644 --- a/src/mavedb/lib/csv/specs.py +++ b/src/mavedb/lib/csv/specs.py @@ -109,13 +109,31 @@ def resolver(self, column_key: str) -> Optional[Callable]: return _optional(lambda data: data.get(column_key)) +def _safe_hgvs_from_post_mapped(mapping: MappedVariant) -> Optional[str]: + """``get_hgvs_from_post_mapped`` with its raises absorbed, since a cell cannot afford to raise. + + ``hgvs_from_vrs_allele`` raises on two payload shapes: a VRS 1.x object, which it refuses + deliberately, and an allele with no ``expressions`` key. Raised from inside a cell resolver, either + would abort the export of every other row in the file. Returning None keeps the refusal — no HGVS is + emitted for those payloads — without letting one variant's shape cost the caller the whole download. + + Intentionally scoped to the CSV export layer, where a single problematic variant should not abort the entire export. + """ + if not mapping.post_mapped: + return None + try: + return get_hgvs_from_post_mapped(mapping.post_mapped) + except (KeyError, ValueError): + return None + + def _post_mapped_hgvs_g(mapping: Optional[MappedVariant]) -> Optional[str]: """The genomic HGVS expression, falling back to one parsed out of the post-mapped VRS object.""" if mapping is None: return None if mapping.hgvs_g: return str(mapping.hgvs_g) - fallback = get_hgvs_from_post_mapped(mapping.post_mapped) if mapping.post_mapped else None + fallback = _safe_hgvs_from_post_mapped(mapping) return fallback if fallback is not None and is_hgvs_g(fallback) else None @@ -125,7 +143,7 @@ def _post_mapped_hgvs_p(mapping: Optional[MappedVariant]) -> Optional[str]: return None if mapping.hgvs_p: return str(mapping.hgvs_p) - fallback = get_hgvs_from_post_mapped(mapping.post_mapped) if mapping.post_mapped else None + fallback = _safe_hgvs_from_post_mapped(mapping) return fallback if fallback is not None and is_hgvs_p(fallback) else None diff --git a/tests/helpers/mocks/factories.py b/tests/helpers/mocks/factories.py index 4fbc5bfa3..6e83d089c 100644 --- a/tests/helpers/mocks/factories.py +++ b/tests/helpers/mocks/factories.py @@ -325,6 +325,11 @@ def create_mock_mapped_variant( pre_mapped=None, post_mapped=None, variant_data=_UNSET, + hgvs_c=_UNSET, + hgvs_g=_UNSET, + hgvs_p=_UNSET, + hgvs_assay_level=_UNSET, + vep_functional_consequence=_UNSET, ): """Create a mock MappedVariant with specified properties. @@ -332,6 +337,16 @@ def create_mock_mapped_variant( variant of a different shape. ``variant_data`` overrides the variant's ``data`` wholesale, which is how a variant with an absent or non-numeric score is expressed — ``score`` alone can only produce a well-formed ``score_data``. + + The ``hgvs_*`` fields and ``vep_functional_consequence`` are read by the CSV surfaces and by nothing + in the annotation layer. All are set explicitly, because an attribute left unset on a MagicMock + resolves to a truthy mock rather than to absent data — which a CSV row would then carry through as a + mock repr instead of a value or NA. + + ``hgvs_g`` and ``hgvs_p`` default to None rather than to a value, deliberately. The CSV resolvers + prefer the stored column and fall back to parsing the post-mapped VRS object, so a populated column + short-circuits the fallback. Since the fallback is the path the variant shape list exists to + exercise, leaving these unset by default is what makes the payload matter. """ mock_variant = create_mock_variant(urn=urn, score=score, score_set=score_set, data=variant_data) @@ -342,6 +357,13 @@ def create_mock_mapped_variant( clingen_allele_id=clingen_allele_id, pre_mapped=TEST_VALID_PRE_MAPPED_VRS_ALLELE_VRS2_X if pre_mapped is None else pre_mapped, post_mapped=TEST_VALID_POST_MAPPED_VRS_ALLELE_VRS2_X if post_mapped is None else post_mapped, + hgvs_c="NM_000271.5:c.3082G>C" if hgvs_c is _UNSET else hgvs_c, + hgvs_g=None if hgvs_g is _UNSET else hgvs_g, + hgvs_p=None if hgvs_p is _UNSET else hgvs_p, + hgvs_assay_level=("NC_000018.10:g.23536836C>G" if hgvs_assay_level is _UNSET else hgvs_assay_level), + vep_functional_consequence=( + "missense_variant" if vep_functional_consequence is _UNSET else vep_functional_consequence + ), ) diff --git a/tests/helpers/variant_shapes.py b/tests/helpers/variant_shapes.py index b9a8ef790..1d60c7e26 100644 --- a/tests/helpers/variant_shapes.py +++ b/tests/helpers/variant_shapes.py @@ -1,14 +1,26 @@ -"""The mapped-variant shapes every annotation surface has to survive. +"""The mapped-variant shapes every export surface has to survive. -Annotation tests otherwise all run against one variant, so a defect that only appears for a particular -stored payload passes the whole suite. Both production failures this list exists to catch were of that -kind: a reference-identical variant whose VRS state is a ``ReferenceLengthExpression`` (84125081) and a -null baseline score stripped on the way out (5c155f4d). +Tests otherwise all run against one variant, so a defect that only appears for a particular stored +payload passes the whole suite. Both production failures this list exists to catch were of that kind: a +reference-identical variant whose VRS state is a ``ReferenceLengthExpression`` (84125081) and a null +baseline score stripped on the way out (5c155f4d). -Lives here rather than under ``tests/lib/annotation`` because the CSV surfaces will consume the same -list, and a shape list inside the annotation package would have to move when they do. +Consumed by both the annotation surfaces (``tests/lib/annotation/test_conformance.py``) and the CSV +composer (``tests/lib/csv/test_columns.py``), which is why it lives here rather than in either package. -Adding a shape is one entry in ``VARIANT_SHAPES``. It applies to every annotation surface at once. +Adding a shape is one entry in ``VARIANT_SHAPES``, and it applies to every surface at once. Not every +shape bears on every surface: ``unmapped_hgvs_columns`` changes only CSV output, and the calibration +shapes change only annotation output. That is fine — a shape that is inert on one surface costs one +cheap assertion there and earns its place on the other. + +Two things are deliberately not shapes here: + +- **gnomAD records and ClinVar controls.** The CSV composer does read both, but as separate per-row + arguments rather than as properties of a mapped variant, so varying them is an orthogonal axis. The + annotation layer reads neither. +- **``score_data`` with no ``score`` key.** Score dataframes are rejected without a ``score`` column + (``lib/validation/dataframe``), so a stored variant always has the key; it may be null, which is + ``null_score``. Expected to be short-lived. This builds on the non-DB mock factories in ``tests/helpers/mocks/factories.py``, hand-rolling the override plumbing — ``kwargs`` forwarded to a factory, plus a ``mutate`` hook for the axes @@ -168,6 +180,11 @@ def _make_non_coding(mapped_variant) -> None: why="a regulatory target, whose identifier comes from post-mapped metadata", mutate=_make_non_coding, ), + VariantShape( + name="unmapped_hgvs_columns", + why="a mapping carrying no hgvs_c, assay-level hgvs, or VEP consequence; CSV-only axis", + kwargs={"hgvs_c": None, "hgvs_assay_level": None, "vep_functional_consequence": None}, + ), ] diff --git a/tests/lib/csv/test_columns.py b/tests/lib/csv/test_columns.py index 429dfe718..35c1ba889 100644 --- a/tests/lib/csv/test_columns.py +++ b/tests/lib/csv/test_columns.py @@ -1,3 +1,6 @@ +import csv +from io import StringIO + import pytest from mavedb.lib.annotation.flatten import FlatAnnotation @@ -9,7 +12,10 @@ rows_to_csv, variant_to_csv_row, ) +from mavedb.lib.csv.namespaces import CsvNamespace from tests.helpers.constants import VALID_CALIBRATION_URN +from tests.helpers.mocks.factories import create_mock_mapped_variant +from tests.helpers.variant_shapes import VARIANT_SHAPES, shape_ids # --------------------------------------------------------------------------- # MockVariant @@ -558,3 +564,88 @@ def test_scores_and_custom_scores_share_a_prefix_without_colliding(self): def test_a_namespace_sharing_a_prefix_still_collides_on_a_repeated_column(self): with pytest.raises(ValueError, match="duplicate columns"): assemble_csv_headers({"scores": ["score"], "scores_custom": ["score"]}, namespaced=True) + + +# --------------------------------------------------------------------------- +# TestCsvRowAcrossVariantShapes +# --------------------------------------------------------------------------- + + +class TestCsvRowAcrossVariantShapes: + """Compose a row for every mapped-variant shape the export surfaces have to survive. + + The tests above specify the row machinery against purpose-built inputs. These run the same composer + over the shared shape list, which is where payload-dependent breakage lives: the ``mavedb`` namespace + resolves ``post_mapped_hgvs_g``, ``post_mapped_hgvs_p``, and ``post_mapped_vrs_digest`` by walking the + stored VRS object, and that object takes every form in the list — VRS 1.x nesting, a cis-phased block, + the three state types, and an allele carrying no expressions at all. + + The contract is deliberately narrow, matching the annotation conformance suite: composing a row never + raises, and the row carries exactly the planned columns. What each value should *be* is specified per + namespace above, not re-asserted per shape. + """ + + # Every namespace whose resolvers read the variant or its mapping. gnomAD and ClinVar are excluded: + # they are separate per-row arguments rather than properties of a shape, so they vary independently. + SHAPE_SENSITIVE_NAMESPACES = [ + CsvNamespace.REFERENCE_HGVS, + CsvNamespace.SCORES, + CsvNamespace.VEP, + CsvNamespace.CLINGEN, + ] + + DATASET_COLUMNS = {"score_columns": ["score"], "count_columns": []} + + def _plan(self): + return plan_csv_columns(self.DATASET_COLUMNS, [str(ns) for ns in self.SHAPE_SENSITIVE_NAMESPACES]) + + @pytest.mark.parametrize("shape", VARIANT_SHAPES, ids=shape_ids()) + def test_a_row_composes_and_carries_every_planned_column(self, shape): + mapped_variant = shape.build(create_mock_mapped_variant) + plan = self._plan() + + row = variant_to_csv_row(mapped_variant.variant, plan.namespaced_columns, mapping=mapped_variant) + + assert set(row) == set(assemble_csv_headers(plan.namespaced_columns)) + + @pytest.mark.parametrize("shape", VARIANT_SHAPES, ids=shape_ids()) + def test_no_cell_leaks_a_mock(self, shape): + """Guards the fixtures rather than the code, and is here because it caught a real mistake. + + ``_value_or_na`` stringifies whatever it is handed, so a resolver reading a MappedVariant field + the factory never set gets a truthy MagicMock and writes its repr into the CSV as a perfectly + well-formed string. Asserting cells are strings does not catch that; asserting they are not mocks + does. Every shape here would have passed a type check while carrying ```` in three + columns. + """ + mapped_variant = shape.build(create_mock_mapped_variant) + plan = self._plan() + + row = variant_to_csv_row(mapped_variant.variant, plan.namespaced_columns, mapping=mapped_variant) + + leaked = {column: value for column, value in row.items() if "Mock" in str(value)} + assert not leaked, f"{shape.name} leaked mock reprs into the row: {leaked}" + + def test_a_mapping_without_hgvs_columns_renders_them_na(self): + """The CSV-only axis: fields absent on the mapping must render NA, not a stand-in.""" + shape = next(s for s in VARIANT_SHAPES if s.name == "unmapped_hgvs_columns") + mapped_variant = shape.build(create_mock_mapped_variant) + plan = self._plan() + + row = variant_to_csv_row(mapped_variant.variant, plan.namespaced_columns, mapping=mapped_variant) + + assert row["post_mapped_hgvs_c"] == "NA" + assert row["post_mapped_hgvs_at_assay_level"] == "NA" + assert row["vep_functional_consequence"] == "NA" + + @pytest.mark.parametrize("shape", VARIANT_SHAPES, ids=shape_ids()) + def test_the_row_serializes_to_csv(self, shape): + """The row is only useful if it survives the writer; a stray newline would split the record.""" + mapped_variant = shape.build(create_mock_mapped_variant) + plan = self._plan() + columns = assemble_csv_headers(plan.namespaced_columns) + + row = variant_to_csv_row(mapped_variant.variant, plan.namespaced_columns, mapping=mapped_variant) + rendered = rows_to_csv([row], columns) + + assert len(list(csv.reader(StringIO(rendered)))) == 2 From 59213019ba0d607d4fd2073840d3cad1e9515b90 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Tue, 11 Aug 2026 09:02:06 -0700 Subject: [PATCH 6/7] feat(scripts): sweep score-set CSV composition alongside the annotation surfaces CSV export is a reviewer-facing surface and had never been run across the corpus. It shares its per-row composer with the variant-level CSV, so sweeping it covers the resolvers where every shape-dependent defect so far has lived -- including the one 53a7b6ee fixed. One unit of work per score set rather than per variant, since the composer builds the file in a single call, with namespaces from discovery so each score set is exercised over exactly what it can emit. Two checks: composition must not raise, and the output must re-parse to a rectangle of the expected size, which catches a value carrying a delimiter or newline that silently splits a record. On a raise, start and limit bisect to the offending row so the report names a variant rather than a file. The CSV surface is sized by total variants, not by current mapped variants, and is attempted even when a score set has no current mappings: the composer emits a row per variant and outer-joins the mapping, so an unmapped variant still gets a row of NA columns. 2309 of 2850 published score sets have more variants than current mappings, so conflating the two counts would report a false row-count violation across most of the corpus. First full run found 26 published score sets whose CSV emits roughly twice the expected rows: 44,218 variants carry two mapped variants both flagged current, one a legacy record whose flag was never cleared when its replacement was written. The annotation surfaces duplicate those variants too, but count mapped variants and so cannot see it. Data repair tracked separately; the export is deliberately not made to dedupe, which would hide a live problem behind correct-looking output. --- src/mavedb/scripts/export_sweep.py | 183 +++++++++++++++++++++++++---- 1 file changed, 159 insertions(+), 24 deletions(-) diff --git a/src/mavedb/scripts/export_sweep.py b/src/mavedb/scripts/export_sweep.py index aa96260ee..792469850 100644 --- a/src/mavedb/scripts/export_sweep.py +++ b/src/mavedb/scripts/export_sweep.py @@ -14,10 +14,9 @@ Writes one CSV row per attempted (score set, surface) pair, successes included: a report showing only failures cannot distinguish "nothing broke" from "nothing ran". Exits non-zero if any pair failed. -Every current mapped variant of every published score set is attempted. There is deliberately no -per-score-set sampling: a sampled sweep can only ever report a score set as unbroken-where-sampled, -and the measured cost of doing all of it is well under an hour, which is the right order for a -pre-release or periodic check. +Every current mapped variant of every published score set is attempted. The measured cost of sweeping all +score set level surfaces at the current database size is well under an hour for CSV surfaces and a little +over an hour for VA annotation surfaces. The sweep is read-only and runs as an anonymous principal, matching what a public consumer receives. Private score calibrations are therefore not exercised; a surface reachable only through a privileged @@ -31,10 +30,11 @@ from dataclasses import dataclass from datetime import datetime, timezone from functools import partial +from io import StringIO from typing import Any, Callable, Optional import asyncclick as click -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.orm import Session from mavedb.lib.annotation.annotate import ( @@ -44,10 +44,14 @@ ) from mavedb.lib.annotation.conformance import AnnotationRoundTripError, round_trip_annotation from mavedb.lib.annotation.exceptions import EXPECTED_ABSENCE_EXCEPTIONS +from mavedb.lib.csv.score_set import available_score_set_csv_namespaces, get_score_set_variants_as_csv from mavedb.lib.permissions.principal import Principal +from mavedb.lib.permissions.score_calibration import ScoreCalibrationViewer from mavedb.lib.score_sets import get_current_mapped_variants_for_annotation +from mavedb.lib.urns import variant_urn_sort_key from mavedb.models.mapped_variant import MappedVariant from mavedb.models.score_set import ScoreSet +from mavedb.models.variant import Variant from mavedb.scripts.environment import script_environment, with_database_session logger = logging.getLogger(__name__) @@ -140,7 +144,7 @@ class SweepTotals: variants_attempted: int = 0 -def sweep_surface( +def sweep_annotation_surface( score_set_urn: str, surface: str, annotate: Callable[[MappedVariant], Optional[Any]], @@ -199,6 +203,121 @@ def published_score_sets(db: Session, max_score_sets: Optional[int]) -> list[Sco return list(db.scalars(query).all()) +def _compose_score_set_csv( + db: Session, score_set: ScoreSet, viewer: ScoreCalibrationViewer, start: Optional[int], limit: Optional[int] +) -> str: + """Compose the score set's CSV over every namespace discovery reports for it. + + Namespaces come from discovery rather than a fixed list, so each score set is exercised over exactly + what it can emit — its own ClinVar releases and calibrations included — which is also what the public + dump does. + """ + namespaces = [entry.namespace for entry in available_score_set_csv_namespaces(db, score_set, viewer)] + return get_score_set_variants_as_csv( + db, score_set, namespaces, namespaced=True, start=start, limit=limit, viewer=viewer + ) + + +def _bisect_to_failing_row( + db: Session, score_set: ScoreSet, viewer: ScoreCalibrationViewer, variant_count: int +) -> Optional[int]: + """The index of the first row whose composition raises, or None if a re-run no longer fails. + + A CSV is composed for the whole score set at once, so a failure names the file rather than the row + that caused it. Halving the window recovers the row in log2(n) recompositions and will only be + executed after a failure has been observed. + """ + low, high = 0, variant_count + while low < high: + middle = (low + high) // 2 + try: + _compose_score_set_csv(db, score_set, viewer, start=low, limit=middle - low + 1) + except Exception: + high = middle + else: + low = middle + 1 + + return low if low < variant_count else None + + +def sweep_csv_surface( + db: Session, score_set: ScoreSet, score_set_urn: str, viewer: ScoreCalibrationViewer, variant_count: int +) -> SurfaceResult: + """Compose the score set's whole variant CSV and check it reads back. + + Unlike the annotation surfaces this is one unit of work per score set, because the composer builds the + file in a single call. It still covers every variant: the same ``variant_to_csv_row`` runs for each + one, which is where the payload-dependent resolvers live. + + Two checks. The composition must not raise — a resolver raising from inside one cell takes the whole + file with it, see commit 53a7b6ee. The output must read back as a rectangle of the expected size, which + catches a value carrying a delimiter or newline that silently splits a record. + """ + result = SurfaceResult(score_set_urn=score_set_urn, surface="score-set-csv", variants_attempted=variant_count) + + try: + csv_text = _compose_score_set_csv(db, score_set, viewer, start=None, limit=None) + except Exception as err: + failing_index = _bisect_to_failing_row(db, score_set, viewer, variant_count) + result.record_failure(_variant_urn_at(db, score_set, failing_index), EXCEPTION, err) + result.variants_failed = 1 + return result + + parsed = list(csv.reader(StringIO(csv_text))) + if not parsed: + result.record_failure("", SCHEMA_VIOLATION, ValueError("composed CSV was empty, with no header row")) + return result + + header, *data_rows = parsed + if len(data_rows) != variant_count: + result.record_failure( + "", + SCHEMA_VIOLATION, + ValueError(f"re-parsed to {len(data_rows)} data row(s) for {variant_count} variant(s)"), + ) + return result + + ragged = next((index for index, row in enumerate(data_rows) if len(row) != len(header)), None) + if ragged is not None: + result.record_failure( + _variant_urn_at(db, score_set, ragged), + SCHEMA_VIOLATION, + ValueError(f"row has {len(data_rows[ragged])} field(s) against a {len(header)}-column header"), + ) + return result + + result.variants_annotated = variant_count + return result + + +def _variant_urn_at(db: Session, score_set: ScoreSet, index: Optional[int]) -> str: + """The URN of the variant the CSV composer would place at *index*, for reporting a located failure.""" + if index is None: + return "" + + urns = sorted( + (urn for urn in db.scalars(select(Variant.urn).where(Variant.score_set_id == score_set.id)).all() if urn), + key=variant_urn_sort_key, + ) + return urns[index] if 0 <= index < len(urns) else "" + + +def total_variant_counts(db: Session) -> dict[Optional[int], int]: + """Variants per score set id, for the whole corpus, in one query. + + The CSV surface is sized by this rather than by the mapped-variant count the annotation surfaces use. + The CSV selects every variant of a score set and outer-joins its mapping, so an unmapped variant still + gets a row with NA columns — 2309 of 2850 published score sets have more variants than current + mappings, so conflating the two would report a false row-count violation for most of the corpus. + + Keyed by ``Optional[int]`` because ``Variant.score_set_id`` is nullable in the model. TODO(#372). + """ + rows = db.execute( + select(Variant.score_set_id, func.count()).select_from(Variant).group_by(Variant.score_set_id) + ).all() + return {score_set_id: count for score_set_id, count in rows} + + @script_environment.command() @click.option( "--max-score-sets", @@ -219,13 +338,15 @@ def export_sweep(db: Session, max_score_sets: Optional[int], output: Optional[st # Matches the public data export: publishing a score set does not publish its calibrations. principal = Principal() + viewer = principal.viewer_for(ScoreCalibrationViewer) surfaces = _surfaces(principal) score_sets = published_score_sets(db, max_score_sets) + variant_counts = total_variant_counts(db) totals = SweepTotals(score_sets_published=len(score_sets)) logger.info( - f"Sweeping {len(score_sets)} published score sets across {len(surfaces)} surfaces, " + f"Sweeping {len(score_sets)} published score sets across {len(surfaces) + 1} surfaces, " "attempting every current mapped variant of each." ) if max_score_sets is not None: @@ -236,29 +357,43 @@ def export_sweep(db: Session, max_score_sets: Optional[int], output: Optional[st for index, score_set in enumerate(score_sets): urn = score_set.urn or f"" mapped_variants = list(get_current_mapped_variants_for_annotation(db, score_set)) - - if not mapped_variants: - totals.score_sets_skipped += 1 - for surface, _ in surfaces: - skipped = SurfaceResult( - score_set_urn=urn, - surface=surface, - outcome=SKIPPED, - skip_reason="no current mapped variants", + variant_count = variant_counts.get(score_set.id, 0) + surface_results: list[SurfaceResult] = [] + + # The annotation surfaces need a current mapping to have anything to say. The CSV surface does + # not: it emits a row per variant and outer-joins the mapping, so an unmapped variant still gets + # a row of NA columns. The two skip on different conditions for that reason. + if mapped_variants: + surface_results.extend( + sweep_annotation_surface(urn, surface, annotate, mapped_variants) for surface, annotate in surfaces + ) + else: + surface_results.extend( + SurfaceResult( + score_set_urn=urn, surface=surface, outcome=SKIPPED, skip_reason="no current mapped variants" ) - rows.append(skipped.as_row()) - continue - - totals.score_sets_attempted += 1 - totals.variants_attempted += len(mapped_variants) + for surface, _ in surfaces + ) + + if variant_count: + surface_results.append(sweep_csv_surface(db, score_set, urn, viewer, variant_count)) + else: + surface_results.append( + SurfaceResult(score_set_urn=urn, surface="score-set-csv", outcome=SKIPPED, skip_reason="no variants") + ) + + if mapped_variants or variant_count: + totals.score_sets_attempted += 1 + totals.variants_attempted += len(mapped_variants) + else: + totals.score_sets_skipped += 1 - for surface, annotate in surfaces: - result = sweep_surface(urn, surface, annotate, mapped_variants) + for result in surface_results: rows.append(result.as_row()) if result.outcome in FAILURE_OUTCOMES: logger.error( - f"{urn} / {surface}: {result.outcome} on {result.variants_failed} of " + f"{urn} / {result.surface}: {result.outcome} on {result.variants_failed} of " f"{result.variants_attempted} variant(s); first was " f"{result.first_failing_variant_urn} ({result.exception_class}: {result.message})" ) From 50c3137cbf4c5fd59e3c629d767a706f429afe91 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Tue, 11 Aug 2026 12:43:46 -0700 Subject: [PATCH 7/7] fix(tests): import annotation utils from optional conftest utility --- tests/lib/annotation/conftest.py | 12 ------------ tests/lib/annotation/conftest_optional.py | 12 ++++++++++++ 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/tests/lib/annotation/conftest.py b/tests/lib/annotation/conftest.py index d93e66c4b..29a056c67 100644 --- a/tests/lib/annotation/conftest.py +++ b/tests/lib/annotation/conftest.py @@ -9,7 +9,6 @@ import pytest -from mavedb.lib.annotation.util import CALIBRATION_SCOPE_EXTENSION_NAME from tests.helpers.constants import PRIVATE_CALIBRATION_OWNER_ID from tests.helpers.mocks.factories import ( create_mock_mapped_variant, @@ -24,17 +23,6 @@ pass -def scope_of(annotation) -> str: - """The disclosed principal of an annotation, which every emitted object must carry.""" - scopes = [ - extension.value - for extension in (annotation.extensions or []) - if extension.name == CALIBRATION_SCOPE_EXTENSION_NAME - ] - assert len(scopes) == 1, f"expected exactly one calibration scope extension, found {scopes}" - return scopes[0] - - def make_private(mapped_variant, *, owner_id: int = PRIVATE_CALIBRATION_OWNER_ID): """Mark every calibration on a mapped variant's score set private, owned by ``owner_id``. diff --git a/tests/lib/annotation/conftest_optional.py b/tests/lib/annotation/conftest_optional.py index 9b5365566..143317ad4 100644 --- a/tests/lib/annotation/conftest_optional.py +++ b/tests/lib/annotation/conftest_optional.py @@ -1,5 +1,6 @@ from unittest.mock import Mock +from mavedb.lib.annotation.util import CALIBRATION_SCOPE_EXTENSION_NAME from mavedb.lib.permissions.principal import Principal from mavedb.models.enums.user_role import UserRole from tests.helpers.constants import PRIVATE_CALIBRATION_OWNER_ID @@ -11,3 +12,14 @@ def admin_principal() -> Principal: def owner_principal(owner_id: int = PRIVATE_CALIBRATION_OWNER_ID) -> Principal: return Principal(Mock(user=Mock(id=owner_id, username="owner"), active_roles=[])) + + +def scope_of(annotation) -> str: + """The disclosed principal of an annotation, which every emitted object must carry.""" + scopes = [ + extension.value + for extension in (annotation.extensions or []) + if extension.name == CALIBRATION_SCOPE_EXTENSION_NAME + ] + assert len(scopes) == 1, f"expected exactly one calibration scope extension, found {scopes}" + return scopes[0]