| from pathlib import Path |
|
|
| from fastapi.testclient import TestClient |
|
|
| from api import store |
| from api.main import app |
|
|
|
|
| def wait_result(client, job_id): |
| job = client.get(f"/jobs/{job_id}").json() |
| assert job["status"] == "succeeded", job |
| return job["output"] |
|
|
|
|
| def test_complete_endpoint_contract(tmp_path: Path): |
| store.DB_PATH = tmp_path / "test.db" |
| store.init_db() |
| with TestClient(app) as client: |
| assert client.get("/health").status_code == 200 |
| project = client.post( |
| "/projects", |
| json={"name": "Y220C benchmark", "cancer_type": "Lung Cancer"}, |
| ).json() |
| pid = project["project_id"] |
|
|
| ingest = client.post( |
| "/variants:ingest", |
| json={ |
| "project_id": pid, |
| "records": [{ |
| "Site": "Exon 6", "Variant": "TP53:p.Y220C", |
| "Allele Frequency": 0.01, "Type": "Missense", |
| "Effect": "Loss-of-function", |
| }], |
| }, |
| ) |
| assert ingest.status_code == 202 |
| ingest_out = wait_result(client, ingest.json()["job_id"]) |
| variant_id = ingest_out["variant_ids"][0] |
|
|
| analyze = client.post( |
| f"/variants/{variant_id}/analyze", json={"project_id": pid} |
| ) |
| analyzed = wait_result(client, analyze.json()["job_id"]) |
| assert analyzed["hgvs_p"] == "p.Y220C" |
| assert "Docking + MD" in analyzed["route"] |
|
|
| prep = client.post( |
| "/structures:prepare", |
| json={"project_id": pid, "variant_ids": [variant_id]}, |
| ) |
| prep_out = wait_result(client, prep.json()["job_id"]) |
| structure_id = prep_out["structure_ids"][0] |
| assert prep_out["structures"][0]["ddg_kcal"] == 3.78 |
|
|
| pocket = client.post( |
| "/pockets:detect", |
| json={"project_id": pid, "structure_ids": [structure_id]}, |
| ) |
| pocket_out = wait_result(client, pocket.json()["job_id"]) |
| assert pocket_out["pockets"][0]["mutant_created"] is True |
|
|
| screen = client.post( |
| "/compounds:screen", |
| json={"project_id": pid, "variant_id": variant_id, "limit": 10}, |
| ) |
| screen_out = wait_result(client, screen.json()["job_id"]) |
| assert len(screen_out["candidates"]) == 10 |
|
|
| compounds = [{ |
| "compound_id": "CMP-1", "name": "Candidate 1", |
| "mw": 330, "logp": 2.5, "tpsa": 70, "rotbonds": 4, |
| }] |
| dock = client.post( |
| "/docking:run", |
| json={"project_id": pid, "variant_id": variant_id, "compounds": compounds}, |
| ) |
| dock_out = wait_result(client, dock.json()["job_id"]) |
| pose = dock_out["poses"][0] |
| assert pose["units"] == "kcal/mol" |
|
|
| md = client.post( |
| "/md:run", |
| json={ |
| "project_id": pid, |
| "complexes": [{ |
| "complex_id": pose["docking_id"], "dock_mut": pose["dock_mut"], |
| "mutant_preference": pose["mutant_preference"], "rotbonds": 4, |
| }], |
| }, |
| ) |
| md_out = wait_result(client, md.json()["job_id"]) |
| assert md_out["results"][0]["replicas"] == 3 |
|
|
| candidate = { |
| "variant": variant_id, "name": "Candidate 1", |
| "Bmut": 0.8, "Sselectivity": 0.7, "MDstability": 0.75, |
| "Frescue": 0.8, "ADMET": 0.7, "Evidence": 0.6, "Risk": 0.2, |
| "confidence": 0.8, "admet_flag": "Acceptable", |
| } |
| ranking = client.post( |
| "/rankings:compute", |
| json={"project_id": pid, "variant_id": variant_id, "candidates": [candidate]}, |
| ) |
| rank_out = wait_result(client, ranking.json()["job_id"]) |
| assert rank_out["shortlist"][0]["rank"] == 1 |
|
|
| report = client.get(f"/projects/{pid}/report") |
| assert report.status_code == 200 |
| body = report.json() |
| assert body["summary"]["variants"] == 1 |
| assert body["summary"]["ranked_candidates"] == 1 |
| assert len(client.get(f"/projects/{pid}/jobs").json()) == 8 |
| assert len(client.get(f"/projects/{pid}/artifacts").json()) >= 6 |
| artifact_id = body["artifacts"][0]["artifact_id"] |
| assert client.get(f"/artifacts/{artifact_id}").status_code == 200 |
|
|
