Kemal Yaylali
ae58e33fe2
Makes a real annotation runnable locally without the 25 GB VEP cache, which is what the demo needs and what a reviewer can reproduce in minutes. - params.vep_database (VEP_DATABASE=true) queries Ensembl's public database instead of a local cache. Slower per variant and fewer fields, so --everything is swapped for the flags the loader actually stores. Its cache placeholder is NO_CACHE, not NO_FILE: Nextflow rejects two staged inputs sharing a filename. - PIPELINE_DATABASE_URL is handed to the pipeline when set. The loader runs inside a container, where the API's own localhost URL would point at the container itself. - README: how to run the UI's annotate button locally against host Nextflow + Docker. Verified end to end on pipeline/tests/data/tiny.vcf: bcftools norm split the multiallelic record, VEP 113 annotated 4 variants live, the loader wrote them and marked the job succeeded, and the UI shows them. The deletion came back as 22:42126611 CT>C with exact VCF alleles, which is the case the audit's ID-tagging fix exists for. Tests: api 51, loader 16, stub run 3/3; ruff, mypy clean.
152 lines
5.4 KiB
Python
152 lines
5.4 KiB
Python
import uuid
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.config import settings
|
|
from app.services import events
|
|
|
|
|
|
async def new_sample(client: AsyncClient) -> str:
|
|
r = await client.post(
|
|
"/api/samples",
|
|
json={"name": f"s-{uuid.uuid4()}", "vcf_uri": "gs://bucket/x.vcf.gz", "assembly": "GRCh37"},
|
|
)
|
|
assert r.status_code == 201, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
@pytest.mark.usefixtures("db")
|
|
async def test_local_without_nextflow_fails_fast_with_instructions(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(settings, "pubsub_topic", None)
|
|
monkeypatch.setattr(events.shutil, "which", lambda _: None)
|
|
sample_id = await new_sample(client)
|
|
|
|
r = await client.post(f"/api/samples/{sample_id}/annotate")
|
|
assert r.status_code == 202
|
|
job = r.json()
|
|
assert job["status"] == "failed"
|
|
assert "make annotate" in job["log"]
|
|
assert job["finished_at"] is not None
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self, returncode: int, stderr: bytes) -> None:
|
|
self.returncode = returncode
|
|
self._stderr = stderr
|
|
self.pid = 4242
|
|
|
|
async def communicate(self) -> tuple[bytes, bytes]:
|
|
return b"", self._stderr
|
|
|
|
|
|
@pytest.mark.usefixtures("db")
|
|
async def test_local_run_is_watched_and_a_crash_marks_the_job_failed(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(settings, "pubsub_topic", None)
|
|
monkeypatch.setattr(events.shutil, "which", lambda _: "/usr/bin/nextflow")
|
|
launched: dict[str, Any] = {}
|
|
|
|
async def fake_exec(*cmd: str, **kw: Any) -> FakeProcess:
|
|
launched["cmd"], launched["env"] = cmd, kw["env"]
|
|
return FakeProcess(1, b"ERROR ~ VEP cache not found")
|
|
|
|
monkeypatch.setattr(events.asyncio, "create_subprocess_exec", fake_exec)
|
|
sample_id = await new_sample(client)
|
|
|
|
r = await client.post(f"/api/samples/{sample_id}/annotate")
|
|
assert r.status_code == 202
|
|
await events.drain()
|
|
|
|
job = (await client.get(f"/api/jobs/{r.json()['id']}")).json()
|
|
assert job["status"] == "failed"
|
|
assert "VEP cache not found" in job["log"]
|
|
# The DB password travels in the environment, never on the command line.
|
|
assert settings.database_url not in " ".join(launched["cmd"])
|
|
assert launched["env"]["DATABASE_URL"] == settings.database_url
|
|
assert launched["cmd"][launched["cmd"].index("--assembly") + 1] == "GRCh37"
|
|
|
|
|
|
class FakeFuture:
|
|
def __init__(self, result: str | Exception) -> None:
|
|
self._result = result
|
|
|
|
def result(self, timeout: float | None = None) -> str:
|
|
if isinstance(self._result, Exception):
|
|
raise self._result
|
|
return self._result
|
|
|
|
|
|
class FakePublisher:
|
|
def __init__(self, result: str | Exception) -> None:
|
|
self.result = result
|
|
self.published: list[tuple[str, bytes]] = []
|
|
|
|
def topic_path(self, project: str, topic: str) -> str:
|
|
return f"projects/{project}/topics/{topic}"
|
|
|
|
def publish(self, topic: str, data: bytes) -> FakeFuture:
|
|
self.published.append((topic, data))
|
|
return FakeFuture(self.result)
|
|
|
|
|
|
@pytest.mark.usefixtures("db")
|
|
async def test_pubsub_publishes_to_the_full_topic_path(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
publisher = FakePublisher("msg-123")
|
|
monkeypatch.setattr(settings, "pubsub_topic", "vcf-uploaded")
|
|
monkeypatch.setattr(settings, "gcp_project", "my-proj")
|
|
monkeypatch.setattr(events, "_publisher", lambda: publisher)
|
|
sample_id = await new_sample(client)
|
|
|
|
r = await client.post(f"/api/samples/{sample_id}/annotate")
|
|
job = r.json()
|
|
assert (job["status"], job["workflow_ref"]) == ("running", "pubsub:msg-123")
|
|
[(topic, data)] = publisher.published
|
|
assert topic == "projects/my-proj/topics/vcf-uploaded"
|
|
assert b'"assembly": "GRCh37"' in data and job["id"].encode() in data
|
|
|
|
|
|
@pytest.mark.usefixtures("db")
|
|
async def test_pubsub_failure_marks_the_job_failed(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(settings, "pubsub_topic", "vcf-uploaded")
|
|
monkeypatch.setattr(settings, "gcp_project", "my-proj")
|
|
monkeypatch.setattr(events, "_publisher", lambda: FakePublisher(RuntimeError("403 denied")))
|
|
sample_id = await new_sample(client)
|
|
|
|
job = (await client.post(f"/api/samples/{sample_id}/annotate")).json()
|
|
assert job["status"] == "failed"
|
|
assert "403 denied" in job["log"]
|
|
|
|
|
|
@pytest.mark.usefixtures("db")
|
|
async def test_the_pipeline_gets_its_own_database_url(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The loader runs in a container, where the API's own localhost URL would point at itself."""
|
|
monkeypatch.setattr(settings, "pubsub_topic", None)
|
|
monkeypatch.setattr(settings, "cloudrun_job", None)
|
|
monkeypatch.setattr(
|
|
settings, "pipeline_database_url", "postgresql+asyncpg://u:[email protected]:5432/db"
|
|
)
|
|
monkeypatch.setattr(events.shutil, "which", lambda _: "/usr/bin/nextflow")
|
|
launched: dict[str, Any] = {}
|
|
|
|
async def fake_exec(*cmd: str, **kw: Any) -> FakeProcess:
|
|
launched["env"] = kw["env"]
|
|
return FakeProcess(0, b"")
|
|
|
|
monkeypatch.setattr(events.asyncio, "create_subprocess_exec", fake_exec)
|
|
sample_id = await new_sample(client)
|
|
|
|
await client.post(f"/api/samples/{sample_id}/annotate")
|
|
await events.drain()
|
|
assert launched["env"]["DATABASE_URL"] == "postgresql+asyncpg://u:[email protected]:5432/db"
|