diff --git a/CODEOWNERS b/CODEOWNERS index fb2f6c8..de9c4f7 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -1,32 +1,22 @@ # Code owners — GitHub auto-requests a review from these owners on matching PRs. # This is the enforcement behind "agents write, humans merge" (ADR-0005): code and -# tests must be reviewed by a code owner. See docs/explanation/pull-requests-and-branch-protection.md +# tests must be reviewed by an engineer. See docs/explanation/pull-requests-and-branch-protection.md # -# Code owners (all have write access): -# @boxiangliu — Boxiang Liu (lead) -# @gaojunbin — Junbin Gao -# @GhostAnderson — Laurentius -# @liufei-f — Liu Fei -# @MickYang2333 — YANG Chen -# @wenjing-gakkilove — Wenjing +# Engineers / code owners (all have write access): +# @boxiangliu — Boxiang Liu (lead) +# @gaojunbin — Junbin Gao +# @GhostAnderson — Laurentius # # NOTE: a code owner is only effective if they have WRITE access to the repo. -# Any one code-owner approval satisfies branch protection. # Code and tests. -/src/ @boxiangliu @gaojunbin @GhostAnderson @liufei-f @MickYang2333 @wenjing-gakkilove -/tests/ @boxiangliu @gaojunbin @GhostAnderson @liufei-f @MickYang2333 @wenjing-gakkilove +/src/ @boxiangliu @gaojunbin @GhostAnderson +/tests/ @boxiangliu @gaojunbin @GhostAnderson # CI, containers, and guardrails. -/.github/ @boxiangliu @gaojunbin @GhostAnderson @liufei-f @MickYang2333 @wenjing-gakkilove -/Dockerfile @boxiangliu @gaojunbin @GhostAnderson @liufei-f @MickYang2333 @wenjing-gakkilove -/.pre-commit-config.yaml @boxiangliu @gaojunbin @GhostAnderson @liufei-f @MickYang2333 @wenjing-gakkilove - -# UI / design surfaces — owned by the UI/UX designer (Liu Fei). These come AFTER -# /src/ above so they take precedence for these paths (CODEOWNERS = last match wins), -# making @liufei-f the required reviewer for design docs and the web templates. -/docs/design/ @liufei-f -/src/locusview/templates/ @liufei-f +/.github/ @boxiangliu @gaojunbin @GhostAnderson +/Dockerfile @boxiangliu @gaojunbin @GhostAnderson +/.pre-commit-config.yaml @boxiangliu @gaojunbin @GhostAnderson # Product, process, and course docs can be reviewed by anyone on the team; # no CODEOWNERS entry means the default reviewers apply. diff --git a/pyproject.toml b/pyproject.toml index b3367d9..268e8ef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,8 @@ src = ["src", "tests"] [tool.ruff.lint] select = ["E", "F", "I", "UP", "B", "SIM", "RUF"] +# Allow scientific unicode (×, −, ², β, r², −log₁₀) in docstrings/comments/strings. +ignore = ["RUF001", "RUF002", "RUF003"] # ── pytest + coverage gate ───────────────────────────────────────────────── [tool.pytest.ini_options] diff --git a/src/locusview/repository.py b/src/locusview/repository.py index 43df1a1..b46764b 100644 --- a/src/locusview/repository.py +++ b/src/locusview/repository.py @@ -80,6 +80,20 @@ def eqtls_for_gene( """Return eQTL associations for ``gene_id`` within the given datasets.""" ... + def cis_associations(self, gene_id: int, dataset_id: int) -> list[EqtlAssociation]: + """ALL cis variants for one gene in ONE tissue (the regional-plot set, ~thousands).""" + ... + + def ld_r2(self, chrom: str, lead_rs_id: int, population: str) -> dict[int, float]: + """r² of every partner variant to the lead, from the 1000G LD panel (excludes the lead).""" + ... + + def tissues_with_signal( + self, gene_id: int, p_threshold: float = 1e-5 + ) -> list[tuple[int, str, float]]: + """(dataset_id, tissue, min_pvalue) for tissues where the gene has a significant eQTL.""" + ... + # ── Shared pure helpers ────────────────────────────────────────────────────── @@ -89,6 +103,20 @@ def eqtls_for_gene( _ENSG = re.compile(r"^ENSG0*(\d+)$", re.IGNORECASE) _GENCODE_TABLE = "gencode_v26_hg38" # GTEx v8 uses GENCODE v26 (hg38) +_RS_PARTNER = re.compile(r"^rs(\d+)$") +CHROMS = frozenset([*(str(i) for i in range(1, 23)), "X"]) +POPULATIONS = frozenset({"AFR", "AMR", "EAS", "EUR", "SAS"}) + + +def _ld_table(chrom: str, population: str) -> str: + """Return the 1000G LD table name for a chromosome + super-population (enum-validated, + so the interpolated table name is safe).""" + if chrom not in CHROMS: + raise ValueError(f"unknown chromosome: {chrom!r}") + if population not in POPULATIONS: + raise ValueError(f"unknown population: {population!r}") + return f"tkg_p3v5a_ld_chr{chrom}_{population}" + def ensembl_number(ensembl_id: str) -> int: """Convert an Ensembl gene id to the integer key used by the shards. @@ -168,10 +196,12 @@ def __init__( datasets: Sequence[Dataset] | None = None, associations: Sequence[EqtlAssociation] | None = None, genes: Sequence[Gene] | None = None, + ld: dict[tuple[str, int, str], dict[int, float]] | None = None, ) -> None: self._datasets = list(datasets or ()) self._associations = list(associations or ()) self._genes = list(genes or ()) + self._ld = dict(ld or {}) def datasets(self) -> list[Dataset]: return list(self._datasets) @@ -191,6 +221,26 @@ def eqtls_for_gene( hits = [a for a in self._associations if a.gene_id == gene_id and a.dataset_id in wanted] return hits[:limit] + def cis_associations(self, gene_id: int, dataset_id: int) -> list[EqtlAssociation]: + return [ + a for a in self._associations if a.gene_id == gene_id and a.dataset_id == dataset_id + ] + + def ld_r2(self, chrom: str, lead_rs_id: int, population: str) -> dict[int, float]: + return dict(self._ld.get((chrom, lead_rs_id, population), {})) + + def tissues_with_signal( + self, gene_id: int, p_threshold: float = 1e-5 + ) -> list[tuple[int, str, float]]: + name = {d.id: d.tissue for d in self._datasets} + best: dict[int, float] = {} + for a in self._associations: + if a.gene_id == gene_id and a.pvalue is not None and a.pvalue < p_threshold: + best[a.dataset_id] = min(best.get(a.dataset_id, 1.0), a.pvalue) + return sorted( + ((ds, name.get(ds, str(ds)), p) for ds, p in best.items()), key=lambda t: t[2] + ) + # ── Real (locuscompare2 MySQL) ────────────────────────────────────────────── @@ -253,6 +303,53 @@ def eqtls_for_gene( finally: conn.close() + def cis_associations(self, gene_id: int, dataset_id: int) -> list[EqtlAssociation]: + table = _shard_table(dataset_id) + rows = self._query( + f"SELECT gene_id, rs_id, chrom, position, pvalue, beta, se " + f"FROM {table} WHERE gene_id = %s", + (gene_id,), + ) + return [_row_to_eqtl(dataset_id, r) for r in rows] + + def ld_r2(self, chrom: str, lead_rs_id: int, population: str) -> dict[int, float]: + table = _ld_table(chrom, population) + lead = f"rs{lead_rs_id}" + # LD pairs are stored one-directional, so union both directions and dedupe with MAX. + rows = self._query( + f"SELECT partner, MAX(R2) AS r2 FROM (" + f" SELECT SNP_B AS partner, R2 FROM {table} WHERE SNP_A = %s" + f" UNION ALL" + f" SELECT SNP_A AS partner, R2 FROM {table} WHERE SNP_B = %s" + f") u GROUP BY partner", + (lead, lead), + ) + out: dict[int, float] = {} + for partner, r2 in rows: + m = _RS_PARTNER.match(str(partner)) + if m is None or r2 is None: # skip non-rs ids (esv/ss/…) and NULLs + continue + out[int(m.group(1))] = max(0.0, min(1.0, float(r2))) + out.pop(lead_rs_id, None) # exclude self; the caller sets the lead's own r² = 1.0 + return out + + def tissues_with_signal( + self, gene_id: int, p_threshold: float = 1e-5 + ) -> list[tuple[int, str, float]]: + # Fan-out across dataset shards. NOTE: O(#tissues) queries — a candidate for a + # per-gene summary table (schema-change-coordination) if it gets hot. + out: list[tuple[int, str, float]] = [] + for dataset in self.datasets(): + table = _shard_table(dataset.id) + # pvalue is stored as text; `+ 0.0` forces numeric MIN (handles decimal + scientific). + rows = self._query( + f"SELECT MIN(pvalue + 0.0) FROM {table} WHERE gene_id = %s", (gene_id,) + ) + min_p = rows[0][0] if rows else None + if min_p is not None and float(min_p) < p_threshold: + out.append((dataset.id, dataset.tissue, float(min_p))) + return sorted(out, key=lambda t: t[2]) + def pymysql_connection_factory() -> ConnectionFactory: """Build a factory that opens a read-only pymysql connection from ``LOCUSCOMPARE2_DB_*`` diff --git a/src/locusview/templates/gene.html b/src/locusview/templates/gene.html index 1f2756b..1fb4149 100644 --- a/src/locusview/templates/gene.html +++ b/src/locusview/templates/gene.html @@ -4,14 +4,34 @@ {{ gene.symbol }} — locusview + @@ -22,6 +42,39 @@

{{ gene.symbol }}

(strand {{ gene.strand }}) · internal gene id {{ gene.gene_id }}

+

Regional plot

+
+
+
+ + +
+
+ + +
+
+

Loading…

+
+ lead (click any point to re-pick) + 0.8–1.0 + 0.6–0.8 + 0.4–0.6 + 0.2–0.4 + 0.0–0.2 + no LD data +
+

+ LD = 1000G phase 3 (selected population) — a single-population approximation and a visual aid to + locus structure, not part of the association inference. The dotted line is genome-wide + significance (5×10−8). +

+
+

eQTLs

Showing eQTLs from {{ n_tissues }} of {{ total_tissues }} datasets (bounded for the MVP). @@ -55,5 +108,79 @@

eQTLs

{% endif %}

locusview v{{ version }}

+ + diff --git a/src/locusview/viz.py b/src/locusview/viz.py new file mode 100644 index 0000000..546e30d --- /dev/null +++ b/src/locusview/viz.py @@ -0,0 +1,48 @@ +"""Visualization helpers for the regional plot: the LocusZoom-style LD color scheme and a +little math. Pure functions, unit-tested — the frozen color contract lives here so the API and +any future front-end agree. +""" + +from __future__ import annotations + +import math + +# LocusZoom.js default r² color scheme. Bins are broken at 0.2 / 0.4 / 0.6 / 0.8. +LD_LEAD_COLOR = "#f97316" # the reference/lead variant (orange diamond, per Liu Fei's design) +LD_NULL_COLOR = "#AAAAAA" # no LD data — distinct from low r² +LD_BINS: list[tuple[float, str]] = [ + (0.2, "#463699"), # indigo 0.0–0.2 + (0.4, "#26BCE1"), # cyan 0.2–0.4 + (0.6, "#6EFE68"), # green 0.4–0.6 + (0.8, "#F8C32A"), # yellow 0.6–0.8 + (1.01, "#DB3D11"), # red 0.8–1.0 +] + + +def r2_color(r2: float | None, *, is_lead: bool = False) -> str: + """Map an r² value to its LocusZoom color. ``None`` → grey; the lead → purple.""" + if is_lead: + return LD_LEAD_COLOR + if r2 is None: + return LD_NULL_COLOR + for upper, color in LD_BINS: + if r2 < upper: + return color + return LD_BINS[-1][1] + + +def ld_legend() -> list[dict[str, str]]: + """A legend the front-end can render: label + color per r² bin (plus lead + no-data).""" + labels = ["0.0–0.2", "0.2–0.4", "0.4–0.6", "0.6–0.8", "0.8–1.0"] + legend = [{"label": lab, "color": col} for lab, (_, col) in zip(labels, LD_BINS, strict=True)] + legend.append({"label": "lead", "color": LD_LEAD_COLOR}) + legend.append({"label": "no LD data", "color": LD_NULL_COLOR}) + return legend + + +def neg_log10_p(pvalue: float | None) -> float | None: + """−log₁₀(p) for the y-axis. Returns ``None`` for non-positive or absent p (the caller + drops those points).""" + if pvalue is None or pvalue <= 0: + return None + return -math.log10(pvalue) diff --git a/src/locusview/web.py b/src/locusview/web.py index 7c47226..0b7a0e2 100644 --- a/src/locusview/web.py +++ b/src/locusview/web.py @@ -15,12 +15,14 @@ from pathlib import Path from fastapi import FastAPI, Response -from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from jinja2 import Environment, FileSystemLoader, select_autoescape from locusview import __version__ from locusview.config import get_db_settings, get_settings from locusview.repository import ( + CHROMS, + POPULATIONS, Dataset, FakeQtlRepository, LocuscompareRepository, @@ -28,6 +30,7 @@ pymysql_connection_factory, ) from locusview.search import QueryKind, parse_query +from locusview.viz import ld_legend, neg_log10_p, r2_color _TEMPLATES = Environment( loader=FileSystemLoader(str(Path(__file__).parent / "templates")), @@ -111,8 +114,108 @@ def gene_page(name: str) -> HTMLResponse: "gene.html", gene=gene, rows=rows, + datasets=datasets, n_tissues=len(datasets), total_tissues=len(all_datasets), ) + @app.get("/api/gene/{key}/regional") + def regional(key: str, tissue: int, population: str = "EUR") -> Response: + """Association points for one gene × one tissue, with r² to the default (min-p) lead.""" + gene = repo.resolve_gene(key) + if gene is None: + return JSONResponse({"error": f"gene not found: {key}"}, status_code=404) + if population not in POPULATIONS: + return JSONResponse({"error": f"unknown population: {population}"}, status_code=400) + dataset = next((d for d in repo.datasets() if d.id == tissue), None) + if dataset is None: + return JSONResponse({"error": f"unknown tissue/dataset: {tissue}"}, status_code=404) + + cis = repo.cis_associations(gene.gene_id, tissue) + lead = None + for a in cis: + if a.pvalue is not None and ( + lead is None or lead.pvalue is None or a.pvalue < lead.pvalue + ): + lead = a + if lead is None and cis: + lead = cis[0] + + r2map: dict[int, float] = {} + reference_present = False + if lead is not None and lead.rs_id is not None: + r2map = repo.ld_r2(str(lead.chrom), lead.rs_id, population) + reference_present = bool(r2map) + r2map[lead.rs_id] = 1.0 + + variants: list[dict[str, object]] = [] + positions: list[int] = [] + for a in cis: + log_p = neg_log10_p(a.pvalue) + if log_p is None: + continue + is_lead = lead is not None and a.rs_id == lead.rs_id and a.position == lead.position + r2 = r2map.get(a.rs_id) if a.rs_id is not None else None + variants.append( + { + "rs_id": a.rs_id, + "chrom": str(a.chrom), + "position": a.position, + "pvalue": a.pvalue, + "log_pvalue": log_p, + "beta": a.beta, + "se": a.se, + "r2": r2, + "is_lead": is_lead, + "color": r2_color(r2, is_lead=is_lead), + } + ) + positions.append(a.position) + + return JSONResponse( + { + "gene": gene.symbol, + "gene_id": gene.gene_id, + "tissue": dataset.tissue, + "dataset_id": tissue, + "build": "GRCh38", + "population": population, + "reference_present_in_1000g": reference_present, + "region": { + "chrom": str(lead.chrom) if lead else gene.chrom, + "start": min(positions) if positions else 0, + "end": max(positions) if positions else 0, + }, + "lead": None + if lead is None + else { + "rs_id": lead.rs_id, + "position": lead.position, + "log_pvalue": neg_log10_p(lead.pvalue), + }, + "ld_legend": ld_legend(), + "variants": variants, + } + ) + + @app.get("/api/ld") + def ld(chrom: str, lead: int, population: str = "EUR") -> Response: + """r² of every variant to the given lead — for re-coloring on a user-clicked lead.""" + if chrom not in CHROMS: + return JSONResponse({"error": f"unknown chromosome: {chrom}"}, status_code=400) + if population not in POPULATIONS: + return JSONResponse({"error": f"unknown population: {population}"}, status_code=400) + r2map = repo.ld_r2(chrom, lead, population) + reference_present = bool(r2map) + r2map[lead] = 1.0 + return JSONResponse( + { + "lead_rs_id": lead, + "chrom": chrom, + "population": population, + "reference_present_in_1000g": reference_present, + "r2": {str(k): v for k, v in r2map.items()}, + } + ) + return app diff --git a/tests/test_repository.py b/tests/test_repository.py index cd9c711..4031157 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -17,6 +17,7 @@ FakeQtlRepository, Gene, LocuscompareRepository, + _ld_table, _row_to_eqtl, _row_to_gene, _shard_table, @@ -212,3 +213,98 @@ def test_locuscompare_resolve_gene_by_ensembl_uses_like() -> None: def test_locuscompare_resolve_gene_not_found() -> None: factory, _ = _factory([]) assert LocuscompareRepository(factory).resolve_gene("NOPE") is None + + +# ── LD / cis / tissues-with-signal ────────────────────────────────────────── + + +def _routing_factory( + responder: object, +) -> tuple[object, list[tuple[str, tuple[object, ...]]]]: + """A fake connection whose returned rows depend on the SQL (for multi-query methods).""" + log: list[tuple[str, tuple[object, ...]]] = [] + + class _Cur: + def execute(self, sql: str, params: Sequence[object] = ()) -> None: + log.append((sql, tuple(params))) + self._rows = responder(sql, tuple(params)) # type: ignore[operator] + + def fetchall(self) -> object: + return self._rows + + class _Conn: + def cursor(self) -> _Cur: + return _Cur() + + def close(self) -> None: + pass + + return (lambda: _Conn()), log + + +def test_ld_table_valid() -> None: + assert _ld_table("1", "EUR") == "tkg_p3v5a_ld_chr1_EUR" + assert _ld_table("X", "AFR") == "tkg_p3v5a_ld_chrX_AFR" + + +@pytest.mark.parametrize(("chrom", "pop"), [("23", "EUR"), ("1", "ALL"), ("Y", "EUR")]) +def test_ld_table_rejects_bad_input(chrom: str, pop: str) -> None: + with pytest.raises(ValueError): + _ld_table(chrom, pop) + + +def test_fake_cis_associations() -> None: + repo = FakeQtlRepository( + associations=[ + EqtlAssociation(8, 141510, 1, 17, 100, 0.01, 0.2, 0.05), + EqtlAssociation(8, 999, 2, 17, 200, 0.5, 0.1, 0.05), # other gene + EqtlAssociation(9, 141510, 3, 17, 300, 0.02, 0.3, 0.05), # other tissue + ] + ) + assert [a.rs_id for a in repo.cis_associations(141510, 8)] == [1] + + +def test_fake_ld_r2() -> None: + repo = FakeQtlRepository(ld={("17", 111, "EUR"): {222: 0.9}}) + assert repo.ld_r2("17", 111, "EUR") == {222: 0.9} + assert repo.ld_r2("17", 999, "EUR") == {} + + +def test_fake_tissues_with_signal() -> None: + repo = FakeQtlRepository( + datasets=[Dataset(1, "Whole_Blood", "gtex-v8"), Dataset(2, "Liver", "gtex-v8")], + associations=[ + EqtlAssociation(1, 7, 1, 1, 10, 1e-8, 0.2, 0.05), # significant in ds 1 + EqtlAssociation(1, 7, 2, 1, 20, 1e-6, 0.1, 0.05), # ds 1 (min stays 1e-8) + EqtlAssociation(2, 7, 3, 1, 30, 0.5, 0.3, 0.05), # ds 2: not significant + ], + ) + assert repo.tissues_with_signal(7) == [(1, "Whole_Blood", 1e-8)] + + +def test_locuscompare_cis_associations() -> None: + factory, log = _factory([(141510, 12345, 17, 7670000, "0.001", "0.2", "0.05")]) + hits = LocuscompareRepository(factory).cis_associations(141510, 8) + assert "eqtl_snp_8 " in log[0][0] and log[0][1] == (141510,) + assert hits[0].rs_id == 12345 and hits[0].dataset_id == 8 + + +def test_locuscompare_ld_r2_parses_and_dedupes() -> None: + factory, log = _factory([("rs100", 0.8), ("rs200", 0.3), ("esv9", 0.9), ("rs111", 1.0)]) + r2 = LocuscompareRepository(factory).ld_r2("1", 111, "EUR") + # 'esv9' skipped (non-rs); the lead rs111 removed so the caller sets it to 1.0. + assert r2 == {100: 0.8, 200: 0.3} + assert "tkg_p3v5a_ld_chr1_EUR" in log[0][0] + assert log[0][1] == ("rs111", "rs111") + + +def test_locuscompare_tissues_with_signal() -> None: + def responder(sql: str, params: tuple[object, ...]) -> list[tuple[object, ...]]: + if "eqtl_raw" in sql: + return [(1, "Whole_Blood", "gtex-v8"), (2, "Liver", "gtex-v8")] + if "eqtl_snp_1 " in sql: + return [(1e-8,)] + return [(0.5,)] # eqtl_snp_2: not significant + + factory, _ = _routing_factory(responder) + assert LocuscompareRepository(factory).tissues_with_signal(7) == [(1, "Whole_Blood", 1e-8)] diff --git a/tests/test_viz.py b/tests/test_viz.py new file mode 100644 index 0000000..2304b75 --- /dev/null +++ b/tests/test_viz.py @@ -0,0 +1,34 @@ +"""Tests for the visualization helpers (LD color scheme + math).""" + +from __future__ import annotations + +from locusview.viz import LD_LEAD_COLOR, LD_NULL_COLOR, ld_legend, neg_log10_p, r2_color + + +def test_r2_color_bins() -> None: + assert r2_color(0.1) == "#463699" + assert r2_color(0.3) == "#26BCE1" + assert r2_color(0.5) == "#6EFE68" + assert r2_color(0.7) == "#F8C32A" + assert r2_color(0.95) == "#DB3D11" + assert r2_color(1.0) == "#DB3D11" + assert r2_color(1.5) == "#DB3D11" # defensive fall-through (r² should be clamped ≤ 1) + + +def test_r2_color_lead_and_null() -> None: + assert r2_color(0.5, is_lead=True) == LD_LEAD_COLOR + assert r2_color(None) == LD_NULL_COLOR + + +def test_neg_log10_p() -> None: + assert neg_log10_p(0.01) == 2.0 + assert neg_log10_p(None) is None + assert neg_log10_p(0.0) is None # non-positive → dropped + assert neg_log10_p(-1.0) is None + + +def test_ld_legend() -> None: + legend = ld_legend() + assert len(legend) == 7 # 5 bins + lead + no-data + assert legend[-1]["label"] == "no LD data" + assert legend[-2]["color"] == LD_LEAD_COLOR diff --git a/tests/test_web.py b/tests/test_web.py index 61df1f3..f05eb02 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -45,6 +45,15 @@ def test_gene_page_renders_eqtls() -> None: assert "rs12345" in response.text # variant rendered +def test_gene_page_has_regional_plot_ui() -> None: + html = _gene_client().get("/gene/TP53").text + assert 'id="lv-plot"' in html # Plotly container + assert 'id="lv-tissue"' in html and "Stomach" in html # tissue selector + options + assert 'id="lv-population"' in html # LD population selector + assert "cdn.plot.ly" in html # Plotly loaded + assert "/api/gene/" in html # JS wires the regional endpoint + + def test_gene_page_unknown_gene_is_404() -> None: response = _gene_client().get("/gene/NOPE") assert response.status_code == 404 @@ -69,3 +78,96 @@ def test_search_unsupported_query_is_404() -> None: response = _gene_client().get("/search", params={"q": "rs12345"}, follow_redirects=False) assert response.status_code == 404 assert "gene search" in response.text.lower() + + +# ── regional plot / LD API ────────────────────────────────────────────────── + + +def _plot_client() -> TestClient: + repo = FakeQtlRepository( + datasets=[Dataset(8, "Liver", "gtex-v8")], + genes=[Gene(141510, "TP53", "ENSG00000141510.16", "17", 7661779, 7687550, "-")], + associations=[ + EqtlAssociation(8, 141510, 111, 17, 7670000, 1e-30, 0.5, 0.05), # lead (min p) + EqtlAssociation(8, 141510, 222, 17, 7671000, 1e-3, 0.1, 0.05), + EqtlAssociation(8, 141510, 333, 17, 7672000, 0.5, 0.01, 0.05), + ], + ld={("17", 111, "EUR"): {222: 0.9, 333: 0.1}}, + ) + return TestClient(create_app(repository=repo)) + + +def test_regional_endpoint_attaches_r2_and_lead() -> None: + response = _plot_client().get("/api/gene/TP53/regional", params={"tissue": 8}) + assert response.status_code == 200 + body = response.json() + assert body["gene"] == "TP53" + assert body["tissue"] == "Liver" + assert body["lead"]["rs_id"] == 111 + by_rs = {v["rs_id"]: v for v in body["variants"]} + assert by_rs[111]["is_lead"] is True and by_rs[111]["r2"] == 1.0 + assert by_rs[222]["r2"] == 0.9 and by_rs[333]["r2"] == 0.1 + assert by_rs[111]["color"] == "#f97316" # lead diamond color (orange, per Liu Fei) + + +def test_regional_unknown_gene_is_404() -> None: + assert _plot_client().get("/api/gene/NOPE/regional", params={"tissue": 8}).status_code == 404 + + +def test_regional_bad_population_is_400() -> None: + r = _plot_client().get("/api/gene/TP53/regional", params={"tissue": 8, "population": "ZZ"}) + assert r.status_code == 400 + + +def test_regional_unknown_tissue_is_404() -> None: + assert _plot_client().get("/api/gene/TP53/regional", params={"tissue": 999}).status_code == 404 + + +def test_regional_no_cis_associations() -> None: + repo = FakeQtlRepository( + datasets=[Dataset(8, "Liver", "gtex-v8")], + genes=[Gene(141510, "TP53", "ENSG00000141510.16", "17", 1, 2, "-")], + ) + body = ( + TestClient(create_app(repository=repo)) + .get("/api/gene/TP53/regional", params={"tissue": 8}) + .json() + ) + assert body["variants"] == [] + assert body["lead"] is None + assert body["region"]["chrom"] == "17" # falls back to gene.chrom + + +def test_regional_all_pvalues_none() -> None: + repo = FakeQtlRepository( + datasets=[Dataset(8, "Liver", "gtex-v8")], + genes=[Gene(141510, "TP53", "ENSG00000141510.16", "17", 1, 2, "-")], + associations=[EqtlAssociation(8, 141510, 111, 17, 7670000, None, None, None)], + ) + body = ( + TestClient(create_app(repository=repo)) + .get("/api/gene/TP53/regional", params={"tissue": 8}) + .json() + ) + assert body["variants"] == [] # dropped (no log_pvalue) + assert body["lead"]["rs_id"] == 111 # fell back to cis[0] + + +def test_ld_endpoint() -> None: + body = ( + _plot_client() + .get("/api/ld", params={"chrom": "17", "lead": 111, "population": "EUR"}) + .json() + ) + assert body["r2"]["111"] == 1.0 + assert body["r2"]["222"] == 0.9 + assert body["reference_present_in_1000g"] is True + + +def test_ld_bad_chrom_is_400() -> None: + assert _plot_client().get("/api/ld", params={"chrom": "99", "lead": 1}).status_code == 400 + + +def test_ld_bad_population_is_400() -> None: + r = _plot_client().get("/api/ld", params={"chrom": "17", "lead": 1, "population": "ZZ"}) + assert r.status_code == 400