diff --git a/src/mavedb/lib/annotation/conformance.py b/src/mavedb/lib/annotation/conformance.py new file mode 100644 index 00000000..851980cb --- /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/src/mavedb/lib/annotation/exceptions.py b/src/mavedb/lib/annotation/exceptions.py index 27511a56..23662f71 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/lib/csv/specs.py b/src/mavedb/lib/csv/specs.py index 88e6d5f1..b309c674 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/src/mavedb/lib/middleware/__init__.py b/src/mavedb/lib/middleware/__init__.py new file mode 100644 index 00000000..04e9e2b9 --- /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 00000000..871eac5d --- /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/routers/score_sets.py b/src/mavedb/routers/score_sets.py index de0cda70..843db6f1 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 @@ -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, @@ -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,47 @@ 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 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 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( + 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 +1318,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 +1361,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 +1370,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 +1405,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 +1416,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 +1519,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 +1530,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 +1627,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 +1638,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/src/mavedb/scripts/export_sweep.py b/src/mavedb/scripts/export_sweep.py new file mode 100644 index 00000000..79246985 --- /dev/null +++ b/src/mavedb/scripts/export_sweep.py @@ -0,0 +1,434 @@ +""" +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. 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 +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 io import StringIO +from typing import Any, Callable, Optional + +import asyncclick as click +from sqlalchemy import func, 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.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__) + +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_annotation_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()) + + +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", + 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() + 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) + 1} 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)) + 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" + ) + 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 result in surface_results: + rows.append(result.as_row()) + + if result.outcome in FAILURE_OUTCOMES: + logger.error( + 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})" + ) + + # 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() diff --git a/src/mavedb/server_main.py b/src/mavedb/server_main.py index 15071488..880bfcfe 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/helpers/constants.py b/tests/helpers/constants.py index d582312b..faabe6d1 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 76b01194..6e83d089 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,48 @@ def create_mock_mapped_variant( mapped_date=None, clingen_allele_id=None, score_set=None, + 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.""" - 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``. + + 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) 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, + 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 new file mode 100644 index 00000000..1d60c7e2 --- /dev/null +++ b/tests/helpers/variant_shapes.py @@ -0,0 +1,193 @@ +"""The mapped-variant shapes every export surface has to survive. + +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). + +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``, 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 +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, + ), + 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}, + ), +] + + +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_optional.py b/tests/lib/annotation/conftest_optional.py index 9b536556..143317ad 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] diff --git a/tests/lib/annotation/test_annotate.py b/tests/lib/annotation/test_annotate.py index 0d0b091f..cbfc19f2 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 00000000..ed83878f --- /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" diff --git a/tests/lib/csv/test_columns.py b/tests/lib/csv/test_columns.py index 429dfe71..35c1ba88 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 diff --git a/tests/lib/middleware/__init__.py b/tests/lib/middleware/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/lib/middleware/test_errors.py b/tests/lib/middleware/test_errors.py new file mode 100644 index 00000000..76a14b0e --- /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() diff --git a/tests/routers/test_score_set.py b/tests/routers/test_score_set.py index f58aa940..26a45bb3 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 ########################################################################################################################