Kemal Yaylali
11fb6b3d73
An end-to-end audit found the repo could not build, test or run as shipped. This fixes every finding, then adds a Cloud Run track so the demo costs about £1/month idle instead of ~£150. CI (red on its first run) - api: setuptools could not build the package (flat layout with app/ and alembic/) - web: missing @types/node; `vitest run` exited 1 with no test files - pipeline: the stub run needed a gitignored VCF, and no process had a stub block - ruff pinned, mypy configured, DB tests on real Postgres (pgserver locally, service in CI) ML serving (scores were meaningless) - the registered model now carries its own feature engineering and returns predict_proba, so serving sends raw columns and cannot drift from training - resolve by registry alias (stages are deprecated in MLflow 3) and record the real version; re-scoring upserts instead of failing on the unique constraint - ClinVar labels parsed from VEP's lowercase terms Pipeline - exact ref/alt recovered from a CHROM_POS_REF_ALT VCF ID; loading is idempotent - job status reaches running/failed/succeeded, so the UI stops polling dead jobs - DATABASE_URL travels in the environment or a Nextflow secret, never on a command line - VEP cache and plugins staged as inputs; the gcp profile runs tasks on Google Batch Deployment - the API serves /api (matching the ingress); the web app reads its API URL at runtime - migrations run in an init container under a Postgres advisory lock - terraform: custom VPC shared with Batch, private Cloud SQL, API enablement, Workload Identity bindings, Secret Manager, deletion protection - serverless track, now the default: Cloud Run services scaling to zero, a Cloud Run job for the Nextflow driver, and Neon or Cloud SQL behind one DATABASE_URL secret. GKE and Argo remain, behind -var deploy_kubernetes=true. See docs/cloud.md. Correctness and security - 409 on duplicate sample names, 422 on bad paging, natural chromosome ordering, wider VEP text columns, enum dropped on downgrade, the sample's assembly actually used - vcf_uri restricted to gs:// objects or files under the data root, blocking option injection - CORS restricted to configured origins; `make down` no longer deletes volumes Data - docs/data.md records the peer-reviewed, openly licensed sources (GIAB HG002, ClinVar, gnomAD) with citations and an honest evaluation plan; `make data` fetches a chr22 slice Verified: api 50 tests, ml 18, loader 16, web 12; ruff, mypy, svelte-check, terraform validate and both kustomize overlays clean.
137 lines
5.2 KiB
Python
137 lines
5.2 KiB
Python
import json
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from load_db import engine_for, load, parse_variant_id, read_vep_tab, to_rows, vep_version
|
|
from set_job_status import set_status
|
|
from sqlalchemy import text
|
|
|
|
HEADER = [
|
|
"Uploaded_variation", "Location", "Allele", "Consequence", "IMPACT", "SYMBOL",
|
|
"HGVSc", "HGVSp", "gnomADe_AF", "CLIN_SIG", "CADD_PHRED",
|
|
]
|
|
# Longer than the old VARCHAR(120) column.
|
|
LONG_CLIN_SIG = (
|
|
"conflicting_classifications_of_pathogenicity,uncertain_significance,"
|
|
"likely_benign,benign,likely_pathogenic,pathogenic"
|
|
)
|
|
|
|
|
|
def vep_tab(tmp_path: Path, rows: list[list[str]]) -> Path:
|
|
path = tmp_path / "x.vep.tsv"
|
|
path.write_text("\n".join([
|
|
"## ENSEMBL VARIANT EFFECT PREDICTOR v113.0",
|
|
"## Output produced at 2026-09-11 12:00:00",
|
|
"#" + "\t".join(HEADER),
|
|
*("\t".join(r) for r in rows),
|
|
]) + "\n")
|
|
return path
|
|
|
|
|
|
ROWS = [
|
|
# SNV
|
|
["22_19710700_C_T", "22:19710700", "T", "missense_variant", "MODERATE", "TBX1",
|
|
"ENST1:c.1C>T", "ENSP1:p.Arg1Trp", "0.0001", "pathogenic", "28.1"],
|
|
# Deletion: VEP reports a trimmed allele ("-") and a shifted Location; the ID keeps the VCF truth.
|
|
["22_42126611_CT_C", "22:42126612", "-", "frameshift_variant", "HIGH", "CYP2D6",
|
|
"-", "-", "-", LONG_CLIN_SIG, "-"],
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("uid", "expected"),
|
|
[
|
|
("22_19710700_C_T", ("22", 19710700, "C", "T")),
|
|
("chrUn_KI270742v1_100_A_AT", ("chrUn_KI270742v1", 100, "A", "AT")),
|
|
],
|
|
)
|
|
def test_parse_variant_id(uid: str, expected: tuple) -> None:
|
|
assert parse_variant_id(uid) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("uid", ["rs123", "12345", "22_x_A_G", "."])
|
|
def test_parse_variant_id_rejects_ids_not_set_by_normalise(uid: str) -> None:
|
|
with pytest.raises(ValueError, match="CHROM_POS_REF_ALT"):
|
|
parse_variant_id(uid)
|
|
|
|
|
|
def test_to_rows_takes_alleles_from_the_id_and_nulls_dashes(tmp_path: Path) -> None:
|
|
rows = to_rows(read_vep_tab(vep_tab(tmp_path, ROWS)), job_id="j")
|
|
snv, deletion = rows
|
|
assert (snv["chrom"], snv["pos"], snv["ref"], snv["alt"]) == ("22", 19710700, "C", "T")
|
|
assert (deletion["pos"], deletion["ref"], deletion["alt"]) == (42126611, "CT", "C")
|
|
assert snv["gnomad_af"] == 0.0001 and deletion["gnomad_af"] is None
|
|
assert deletion["hgvsc"] is None and deletion["gene"] == "CYP2D6"
|
|
assert deletion["clinvar_sig"] == LONG_CLIN_SIG
|
|
ann = json.loads(deletion["annotations"])
|
|
assert "-" not in ann.values() and "CADD_PHRED" not in ann
|
|
assert json.loads(snv["annotations"])["CADD_PHRED"] == "28.1"
|
|
|
|
|
|
def test_vep_version(tmp_path: Path) -> None:
|
|
assert vep_version(vep_tab(tmp_path, ROWS)) == "113.0"
|
|
|
|
|
|
def test_engine_for_uses_psycopg_for_any_postgres_url() -> None:
|
|
for url in ("postgresql+asyncpg://u:p@h/db", "postgresql://u:p@h/db"):
|
|
assert engine_for(url).url.drivername == "postgresql+psycopg"
|
|
|
|
|
|
def new_job(engine, status: str = "running") -> str:
|
|
job_id = str(uuid.uuid4())
|
|
with engine.begin() as conn:
|
|
conn.execute(text("INSERT INTO jobs (id, status) VALUES (:id, :s)"), {"id": job_id, "s": status})
|
|
return job_id
|
|
|
|
|
|
def job_and_count(engine, job_id: str) -> tuple:
|
|
with engine.connect() as conn:
|
|
job = conn.execute(text("SELECT status, vep_version, log FROM jobs WHERE id=:id"),
|
|
{"id": job_id}).one()
|
|
n = conn.execute(text("SELECT count(*) FROM variants WHERE job_id=:id"), {"id": job_id}).scalar()
|
|
return (*job, n)
|
|
|
|
|
|
def test_load_is_idempotent_and_marks_job_succeeded(engine, tmp_path: Path) -> None:
|
|
job_id = new_job(engine)
|
|
rows = to_rows(read_vep_tab(vep_tab(tmp_path, ROWS)), job_id=job_id)
|
|
load(engine, job_id, rows, vep="113.0")
|
|
load(engine, job_id, rows, vep="113.0") # a retried task must not duplicate variants
|
|
assert job_and_count(engine, job_id) == ("succeeded", "113.0", None, 2)
|
|
|
|
|
|
def test_load_with_no_variants_still_succeeds(engine, tmp_path: Path) -> None:
|
|
job_id = new_job(engine)
|
|
rows = to_rows(read_vep_tab(vep_tab(tmp_path, [])), job_id=job_id)
|
|
load(engine, job_id, rows, vep="113.0")
|
|
assert job_and_count(engine, job_id) == ("succeeded", "113.0", None, 0)
|
|
|
|
|
|
def test_set_status_failed_records_the_reason(engine) -> None:
|
|
job_id = new_job(engine)
|
|
set_status(engine, job_id, "failed", "workflow annotate-abc Failed")
|
|
status, _, log, _ = job_and_count(engine, job_id)
|
|
assert (status, log) == ("failed", "workflow annotate-abc Failed")
|
|
|
|
|
|
def test_set_status_never_overrides_a_succeeded_job(engine) -> None:
|
|
job_id = new_job(engine, status="succeeded")
|
|
set_status(engine, job_id, "failed", "late exit handler")
|
|
assert job_and_count(engine, job_id)[0] == "succeeded"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected_query"),
|
|
[
|
|
# asyncpg's ssl= becomes libpq's sslmode= for psycopg.
|
|
("postgresql+asyncpg://u:p@h/db?ssl=require", {"sslmode": "require"}),
|
|
("postgresql://u:p@h/db?sslmode=require", {"sslmode": "require"}),
|
|
("postgresql+asyncpg://u:p@h/db", {}),
|
|
],
|
|
)
|
|
def test_engine_for_translates_ssl_options(raw: str, expected_query: dict) -> None:
|
|
url = engine_for(raw).url
|
|
assert url.drivername == "postgresql+psycopg"
|
|
assert dict(url.query) == expected_query
|