From 980ea94de4ad20f4cf05c9cb9031c71e4c27a5da Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Tue, 22 Sep 2026 01:52:06 +0000 Subject: [PATCH] Run a gene set analysis end to end: upload validation and the job Second slice of spec 012. Two layers, neither importing Chainlit, because the awkward parts -- a job that outlasts a chat turn, a submission that succeeds and then fails, a result that must be three sizes for three audiences -- are the parts worth testing, and a browser is not needed to test them. **`upload.py`** caps a file at 20 MB against Chainlit's shipped 500 MB. The host has 4.7 GB free of 88 GB and `~/update-beta-chat.sh` refuses to deploy under 6 GB, so a handful of default-sized uploads would take the chat down *and* block the fix. Shape is checked here rather than by the service, which answers a malformed matrix minutes later through a status field, in R's words. **`job.py`** owns one analysis. `Finished` splits by audience: `for_model` bounded and allow-listed, `links` the capability URL the model must never see, `table_path` the full table for the user. Found while writing it: a one-sample file was told it was not an expression matrix -- it is one, it just cannot be compared against anything -- and `gene_count` returned `-1` for a file too long to finish counting, a sentinel of the same type as a real count, leaving a function whose result gets shown to someone. It is `None` now, and the type says so. **Adversarial review of the finished branch found four more, and three are the same mistake.** *The upload leaked on the likeliest paths.* The group checks sat above the `try`, so a wrong group name or a miscounted label list -- the two mistakes a user actually makes -- returned an error and left their matrix on the disk. The leak was on the paths people take most often. *The loading loop had only the wall-clock bound.* That is the bug I had fixed in `await_result` earlier in the same review, in the same file, twelve lines away. Fixing a loop is not fixing the loops. Both now carry `MAX_POLLS` as well. *`finished` and `failed` were defined on the status objects and used by nobody* -- the loops compared the strings themselves, so "done" had three definitions. One `TERMINAL_STATUSES` now, and `failed` is derived as "terminal and not complete", so a status added later is a failure by default rather than an analysis that silently reports no results. *Result tables were written and never removed.* The upload is deleted the moment it is submitted and then the output was kept forever, at ~2 MB each, on a host with 4.7 GB free. Deleting the input and hoarding the output is not a disk policy. `prune_results` runs before each write, by age then by total size, and only ever touches files this module named. **And the sabotage found a fifth.** Removing the `prune_results` call from `await_result` broke nothing: every pruning test called the function directly, so the directory could have grown forever with the suite green. That it works is not the same as it being called. Sabotage, each against the test written for it: validation back above the `try` fails both deletion tests; removing the prune call fails the new one; unbinding the loading loop hangs, which is the failure mode the bound converts into an error. 67 tests in tests/gsa, ./checks.sh clean. Co-Authored-By: Claude Opus 5 --- src/gsa/client.py | 19 +- src/gsa/job.py | 327 ++++++++++++++++++++++++ src/gsa/upload.py | 145 +++++++++++ tests/gsa/test_gsa_job.py | 474 +++++++++++++++++++++++++++++++++++ tests/gsa/test_gsa_upload.py | 122 +++++++++ 5 files changed, 1084 insertions(+), 3 deletions(-) create mode 100644 src/gsa/job.py create mode 100644 src/gsa/upload.py create mode 100644 tests/gsa/test_gsa_job.py create mode 100644 tests/gsa/test_gsa_upload.py diff --git a/src/gsa/client.py b/src/gsa/client.py index 5531830..451d156 100644 --- a/src/gsa/client.py +++ b/src/gsa/client.py @@ -105,6 +105,11 @@ def _checked(kind: str, value: str) -> str: return value +#: Statuses that mean the service has stopped working on it. Listed once so +#: a loop cannot disagree with the dataclass about what "done" means. +TERMINAL_STATUSES = frozenset({"complete", "failed"}) + + @dataclass(frozen=True) class LoadingStatus: """Progress of `POST /data/load`, which is not instant.""" @@ -116,7 +121,11 @@ class LoadingStatus: @property def finished(self) -> bool: - return self.status in {"complete", "failed"} + return self.status in TERMINAL_STATUSES + + @property + def failed(self) -> bool: + return self.finished and self.status != "complete" @dataclass(frozen=True) @@ -146,11 +155,15 @@ class AnalysisStatus: @property def finished(self) -> bool: - return self.status in {"complete", "failed"} + return self.status in TERMINAL_STATUSES @property def failed(self) -> bool: - return self.status == "failed" + # Anything terminal that is not success. Derived rather than + # `== "failed"`, so a status added to TERMINAL_STATUSES later is + # treated as a failure by default instead of being silently + # reported as a completed analysis with no results. + return self.finished and self.status != "complete" class GsaClient: diff --git a/src/gsa/job.py b/src/gsa/job.py new file mode 100644 index 0000000..b1b6589 --- /dev/null +++ b/src/gsa/job.py @@ -0,0 +1,327 @@ +"""Run one analysis from start to finished file. + +Sits between the client, which knows the service, and the chat, which knows +the user. Nothing here imports Chainlit: the awkward parts -- polling a job +that outlasts a chat turn, a submission that succeeds and then fails, a +result that has to be two different sizes for two different audiences -- +are the parts worth testing, and a browser is not needed to test them. + +**The submission is a receipt.** Measured: `POST /analysis` returned 200 and +the analysis then failed with `CONNECTION_FORCED - broker forced connection +closure`, visible only through `/status`. So `submit_*` returns an ID and +promises nothing, and `await_result` is where success or failure is decided. +""" + +import contextlib +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from pathlib import Path + +from gsa import results as gsa_results +from gsa.client import AnalysisStatus, GsaClient, GsaError +from gsa.upload import Matrix, discard +from util.logging import logging + +logger = logging.getLogger(__name__) + +#: A PADOG run with 1,000 permutations took minutes in the measured run. +#: This is a ceiling on waiting, not an expectation. +DEFAULT_DEADLINE_SECONDS = 30 * 60 +POLL_INTERVAL_SECONDS = 10.0 + +#: A hard ceiling on how many times the service is asked, independent of the +#: interval. +#: +#: Two bounds are needed, and the second one took two attempts. The +#: wall-clock deadline assumes each turn of the loop waits; the moment +#: sleeping does not sleep -- a patched `_sleep`, a zero interval from a +#: caller -- elapsed time never advances and the loop hammers the service. +#: My first fix derived the limit as `deadline / interval`, which computes +#: the bound from the quantity that is degenerate: `poll_interval=0` gave a +#: limit of 1.8 million, so the guard against a runaway loop was itself +#: unbounded in exactly the case it existed for. It left three pytest +#: processes spinning at 100% CPU on a shared host. +#: +#: At the real 10s interval a 30-minute deadline is 180 polls, so this is +#: generous. +MAX_POLLS = 2_000 + +#: The service's own default, and the method the round trip was measured +#: with. Camera is faster; choosing between them wants evidence rather than +#: preference, so it is recorded as an open question in spec 012. +DEFAULT_METHOD = "PADOG" + +ProgressCallback = Callable[[AnalysisStatus], Awaitable[None]] + + +class AnalysisFailedError(GsaError): + """The service accepted the analysis and then could not finish it.""" + + +@dataclass(frozen=True) +class Finished: + """A completed analysis, split by audience. + + Three fields, three destinations, kept apart deliberately: + + for_model bounded, allow-listed, safe to put in a prompt + links the user's Pathway Browser view -- a capability URL + that must not reach the model + table_path the full table, for the user to download + """ + + analysis_id: str + for_model: dict[str, object] + links: list[tuple[str, str]] + table_path: Path + + +async def submit_public_dataset( + client: GsaClient, + *, + resource_id: str, + dataset_id: str, + factor: str, + group1: str, + group2: str, + method: str = DEFAULT_METHOD, + deadline_seconds: float = DEFAULT_DEADLINE_SECONDS, +) -> str: + """Load a public dataset and submit it. Returns an analysis ID. + + The matrix is downloaded here and handed straight to `submit`. It is + 1.2 MB for a small dataset and never leaves this function. + """ + loading_id = await client.load_public_dataset(resource_id, dataset_id) + # Both bounds, same as `await_result`. This loop had only the + # wall-clock one -- the bug I had just fixed in its sibling, in the same + # file, during the same review. Fixing a loop is not fixing the loops. + deadline = time.monotonic() + deadline_seconds + polls = 0 + while True: + status = await client.loading_status(loading_id) + polls += 1 + if status.failed: + raise AnalysisFailedError( + f"loading {dataset_id} failed: {status.description}" + ) + if status.finished: + break + if polls >= MAX_POLLS or time.monotonic() > deadline: + raise AnalysisFailedError(f"loading {dataset_id} did not finish in time") + await _sleep(POLL_INTERVAL_SECONDS) + + summary = await client.dataset_summary(dataset_id) + groups = summary.factors.get(factor) + if not groups: + raise AnalysisFailedError( + f"{dataset_id} has no factor called {factor!r}. " + f"It has: {', '.join(sorted(summary.factors)) or 'none'}." + ) + _check_comparable(groups, factor, group1, group2) + + matrix = await client.download_matrix(dataset_id) + return await client.submit( + method=method, + dataset_name=dataset_id, + dataset_type=summary.type, + matrix=matrix, + samples=summary.samples, + analysis_group=groups, + group1=group1, + group2=group2, + ) + + +async def submit_uploaded_matrix( + client: GsaClient, + *, + matrix: Matrix, + dataset_type: str, + analysis_group: list[str], + group1: str, + group2: str, + method: str = DEFAULT_METHOD, +) -> str: + """Submit a user's own matrix, then delete their file. + + The file goes whether or not the analysis then succeeds: the service has + its own copy by then, and this host does not have room to keep ours. + `dataset_name` is deliberately not the user's filename -- that is one of + the strings the disclosure rules exist to keep out of a prompt. + """ + # Everything, including the validation, inside the `try`. + # + # The checks used to sit above it, so a wrong group name or a + # miscounted label list -- the two mistakes a user is most likely to + # make -- returned an error and left their file on disk. The disk leak + # happened on exactly the paths people take most often. + try: + _check_comparable(analysis_group, "the grouping you gave", group1, group2) + if len(analysis_group) != len(matrix.samples): + raise AnalysisFailedError( + f"You gave {len(analysis_group)} group labels for " + f"{len(matrix.samples)} samples. There must be one label per " + f"sample, in the order the columns appear." + ) + return await client.submit( + method=method, + dataset_name="uploaded", + dataset_type=dataset_type, + matrix=matrix.text, + samples=matrix.samples, + analysis_group=analysis_group, + group1=group1, + group2=group2, + ) + finally: + discard(matrix.path) + + +def _check_comparable(groups: list[str], label: str, group1: str, group2: str) -> None: + distinct = set(groups) + missing = [g for g in (group1, group2) if g not in distinct] + if missing: + raise AnalysisFailedError( + f"{', '.join(missing)} is not a value of {label}. " + f"It has: {', '.join(sorted(distinct))}." + ) + if group1 == group2: + raise AnalysisFailedError("The two groups to compare must be different.") + + +#: How long a written result table is kept, and how much of them in total. +#: +#: The upload is deleted the moment it is submitted, and then the *output* +#: was kept forever -- each table is up to ~2 MB, on a host with 4.7 GB +#: free. Deleting the input and hoarding the output is not a disk policy. +#: +#: The window only has to outlast a user downloading their own results. +RESULT_MAX_AGE_SECONDS = 24 * 60 * 60 +RESULT_DIR_MAX_BYTES = 200 * 1024 * 1024 + + +def prune_results( + out_dir: Path, + *, + max_age_seconds: float = RESULT_MAX_AGE_SECONDS, + max_total_bytes: int = RESULT_DIR_MAX_BYTES, + now: float | None = None, +) -> int: + """Delete old result tables. Returns how many were removed. + + Called before each write, so the directory bounds itself and there is + no cron job to forget. Age first, then oldest-first until the total + fits: age alone leaves a burst of results unbounded, and size alone + keeps one stale file forever on a quiet week. + + Only ever touches files this module wrote, matched by name. A cleanup + that globs a directory it does not own is one misconfiguration away + from deleting something else. + """ + moment = time.time() if now is None else now + ours = sorted( + (path for path in out_dir.glob("reactome-gsa-*.tsv") if path.is_file()), + key=lambda path: path.stat().st_mtime, + ) + + removed = 0 + surviving: list[Path] = [] + for path in ours: + if moment - path.stat().st_mtime > max_age_seconds: + with contextlib.suppress(OSError): + path.unlink() + removed += 1 + else: + surviving.append(path) + + total = sum(path.stat().st_size for path in surviving if path.exists()) + for path in surviving: + if total <= max_total_bytes: + break + try: + size = path.stat().st_size + path.unlink() + except OSError: + continue + total -= size + removed += 1 + + if removed: + logger.info("pruned gsa result tables", extra={"removed": removed}) + return removed + + +async def await_result( + client: GsaClient, + analysis_id: str, + *, + out_dir: Path, + on_progress: ProgressCallback | None = None, + deadline_seconds: float = DEFAULT_DEADLINE_SECONDS, + poll_interval: float = POLL_INTERVAL_SECONDS, +) -> Finished: + """Poll until the analysis finishes, then write the table and return. + + Raises `AnalysisFailedError` when the service says so -- which is the only + place it ever says so. + """ + deadline = time.monotonic() + deadline_seconds + max_polls = min( + MAX_POLLS, max(2, int(deadline_seconds / max(poll_interval, 1.0)) + 2) + ) + polls = 0 + + while True: + status = await client.analysis_status(analysis_id) + polls += 1 + if on_progress is not None: + await on_progress(status) + if status.failed: + raise AnalysisFailedError(status.description or "the analysis failed") + if status.finished: + # Terminal and not failed, so: complete. Asking the status + # object rather than comparing the string again keeps one + # definition of "done" across both loops and the dataclass. + break + if polls >= max_polls or time.monotonic() > deadline: + raise AnalysisFailedError( + f"the analysis did not finish within " + f"{deadline_seconds / 60:.0f} minutes; it may still be " + f"running as {analysis_id}" + ) + await _sleep(poll_interval) + + parsed = gsa_results.parse(await client.result(analysis_id)) + if not parsed.pathways: + # Complete, and yet nothing to report. Better to say so than to + # hand back an empty file and an exact-sounding zero. + raise AnalysisFailedError("the analysis finished but returned no pathway table") + + out_dir.mkdir(parents=True, exist_ok=True) + prune_results(out_dir) + table_path = out_dir / f"reactome-gsa-{analysis_id}.tsv" + table_path.write_text(gsa_results.as_tsv(parsed)) + logger.info( + "gsa result written", + extra={ + "analysis": analysis_id, + "pathways": len(parsed.pathways), + "bytes": table_path.stat().st_size, + }, + ) + + return Finished( + analysis_id=analysis_id, + for_model=gsa_results.for_model(parsed), + links=gsa_results.for_user(parsed), + table_path=table_path, + ) + + +async def _sleep(seconds: float) -> None: + # Indirected so a test can run the poll loop without waiting for it. + import asyncio + + await asyncio.sleep(seconds) diff --git a/src/gsa/upload.py b/src/gsa/upload.py new file mode 100644 index 0000000..c5b87fe --- /dev/null +++ b/src/gsa/upload.py @@ -0,0 +1,145 @@ +"""Accept an expression matrix from a user, or refuse it clearly. + +**Why a cap well below Chainlit's.** `.chainlit/config.toml` ships +`max_size_mb = 500`. The host this runs on has 4.7 GB free of 88 GB, and +`~/update-beta-chat.sh` refuses to deploy under 6 GB, so a handful of +default-sized uploads would take the chat down *and* block the fix. A real +matrix is far smaller: the 16-sample melanoma example is 1.2 MB, and 20 MB +covers a large study comfortably. + +**Why shape is checked before the service sees it.** ReactomeGSA answers a +malformed matrix minutes later, through a status field, with a message +written for someone reading R output. A user who pasted the wrong file +deserves to hear so immediately and in their own terms. +""" + +import contextlib +import os +from dataclasses import dataclass +from pathlib import Path + +MAX_UPLOAD_BYTES_ENV = "GSA_MAX_UPLOAD_BYTES" +DEFAULT_MAX_UPLOAD_BYTES = 20 * 1024 * 1024 + +#: Two columns -- a gene identifier and one sample -- is still a table, and +#: saying "this is not an expression matrix" to someone who uploaded one +#: with a single sample is both wrong and unhelpful. So the structural floor +#: is two columns, and "not enough samples to compare" is a separate, +#: specific refusal below. +MIN_COLUMNS = 2 +MIN_DATA_ROWS = 2 + + +class UploadRejectedError(ValueError): + """The file cannot be analysed, with a reason meant for the user.""" + + +@dataclass(frozen=True) +class Matrix: + """A validated matrix, and what could be learned about it cheaply.""" + + path: Path + size_bytes: int + samples: list[str] + #: `None` when the file was too long to finish counting. Not `-1`: a + #: sentinel of the same type as a real count leaves the function, and + #: the first thing anyone does with a gene count is show it to someone. + #: "unknown" is a fact; "-1 genes" is a bug wearing a number. + gene_count: int | None + + @property + def text(self) -> str: + """The matrix itself. Never logged, never shown, never prompted.""" + return self.path.read_text() + + +def max_upload_bytes() -> int: + raw = os.environ.get(MAX_UPLOAD_BYTES_ENV) + if not raw: + return DEFAULT_MAX_UPLOAD_BYTES + try: + value = int(raw) + except ValueError: + return DEFAULT_MAX_UPLOAD_BYTES + return value if value > 0 else DEFAULT_MAX_UPLOAD_BYTES + + +def _split(line: str) -> list[str]: + # Tab first: it is what the service wants and what every export + # produces. Comma only if there is no tab at all, because a TSV cell can + # legitimately contain a comma and splitting on it would silently + # mangle the header rather than fail. + return line.split("\t") if "\t" in line else line.split(",") + + +def validate(path: Path) -> Matrix: + """Check an uploaded file and describe it, or raise `UploadRejectedError`. + + Reads the header and counts lines; does not hold the whole matrix. + """ + size = path.stat().st_size + limit = max_upload_bytes() + if size > limit: + raise UploadRejectedError( + f"That file is {size / 1e6:.1f} MB and the limit is " + f"{limit / 1e6:.0f} MB. An expression matrix this large is " + f"unusual -- if it is right, it needs to go through " + f"reactome.org/PathwayBrowser rather than the chat." + ) + if size == 0: + raise UploadRejectedError("That file is empty.") + + header: list[str] = [] + rows = 0 + counted = True + with path.open(encoding="utf-8", errors="replace") as handle: + for index, line in enumerate(handle): + if not line.strip(): + continue + if not header: + header = _split(line.rstrip("\n")) + continue + rows += 1 + if index > 5000 and rows > MIN_DATA_ROWS: + # Enough to know it is a matrix. Counting every gene of a + # 20,000-row file to answer "is this a matrix" is work + # nobody asked for. + counted = False + break + + if len(header) < MIN_COLUMNS: + raise UploadRejectedError( + "That does not look like an expression matrix. It needs a header " + "row naming the samples, then one row per gene -- tab- or " + "comma-separated, with at least two samples to compare." + ) + if rows == 0: + raise UploadRejectedError("That file has a header but no data rows.") + + # The first header cell labels the gene column and is often blank -- + # the measured example's header starts with a tab. + samples = [name.strip() for name in header[1:] if name.strip()] + if len(samples) < 2: + raise UploadRejectedError( + "There is only one sample in that file. A gene set analysis " + "compares two groups of samples, so it needs at least two." + ) + + return Matrix( + path=path, + size_bytes=size, + samples=samples, + gene_count=rows if counted else None, + ) + + +def discard(path: Path) -> None: + """Delete an uploaded file. Safe to call twice, and on a missing file. + + Called once the matrix has been submitted, whether or not the analysis + then succeeds: the service has its own copy, and this host has 4.7 GB. + """ + # A file that cannot be deleted must not fail an analysis that has + # already been submitted; the disk check on the next deploy will notice. + with contextlib.suppress(OSError): + path.unlink(missing_ok=True) diff --git a/tests/gsa/test_gsa_job.py b/tests/gsa/test_gsa_job.py new file mode 100644 index 0000000..a8b0cbb --- /dev/null +++ b/tests/gsa/test_gsa_job.py @@ -0,0 +1,474 @@ +"""One analysis, start to finished file. + +A stub client rather than the network: the behaviours worth pinning are a +submission that succeeds and then fails, a poll that must not run forever, +and a result that has to be three different things for three different +audiences. None of those needs the service to be up, and all of them would +be untestable if they only existed inside a Chainlit handler. +""" + +import asyncio +import functools +import json +import time +from pathlib import Path +from typing import Any + +import pytest + +from gsa import job +from gsa.client import AnalysisStatus, DatasetSummary, LoadingStatus +from gsa.upload import Matrix + +FIXTURE = Path(__file__).parent / "result_fixture.json" + + +def asyncio_test(fn: Any) -> Any: + """Run an async test on its own loop. + + This repo has no pytest-asyncio; `tests/agent/test_collection_routing.py` + calls `asyncio.run` inline. Same approach, kept out of the test bodies + so each one reads as the sequence it is testing. + """ + + @functools.wraps(fn) + def wrapper(*args: Any, **kwargs: Any) -> Any: + return asyncio.run(fn(*args, **kwargs)) + + return wrapper + + +class StubClient: + """Records what it was asked, answers what it was told to.""" + + def __init__( + self, + *, + statuses: list[AnalysisStatus] | None = None, + loading: list[LoadingStatus] | None = None, + summary: DatasetSummary | None = None, + ) -> None: + self.statuses = statuses or [AnalysisStatus("complete", "Analysis done", 1.0)] + self.loading = loading or [LoadingStatus("complete", "ready", 1.0, "DS")] + self.summary_value = summary + self.submitted: dict[str, Any] = {} + self.matrix_downloads = 0 + + async def load_public_dataset(self, resource_id: str, dataset_id: str) -> str: + return "load-1" + + async def loading_status(self, loading_id: str) -> LoadingStatus: + return self.loading.pop(0) if len(self.loading) > 1 else self.loading[0] + + async def dataset_summary(self, dataset_id: str) -> DatasetSummary: + assert self.summary_value is not None + return self.summary_value + + async def download_matrix(self, dataset_id: str) -> str: + self.matrix_downloads += 1 + return "\tS1\tS2\nENSG1\t1\t2\n" + + async def submit(self, **kwargs: Any) -> str: + self.submitted = kwargs + return "an-1" + + async def analysis_status(self, analysis_id: str) -> AnalysisStatus: + return self.statuses.pop(0) if len(self.statuses) > 1 else self.statuses[0] + + async def result(self, analysis_id: str) -> dict[str, Any]: + loaded: dict[str, Any] = json.loads(FIXTURE.read_text()) + return loaded + + +SUMMARY = DatasetSummary( + dataset_id="EXAMPLE_MEL_RNA", + title="Melanoma RNA-seq example", + type="rnaseq_counts", + samples=["S1", "S2", "S3", "S4"], + factors={"condition": ["MCM", "MOCK", "MCM", "MOCK"]}, +) + + +@pytest.fixture(autouse=True) +def _no_waiting(monkeypatch: pytest.MonkeyPatch) -> None: + async def instant(_seconds: float) -> None: + return None + + monkeypatch.setattr(job, "_sleep", instant) + + +@asyncio_test +async def test_a_public_dataset_reaches_submission(tmp_path: Path) -> None: + client = StubClient(summary=SUMMARY) + analysis_id = await job.submit_public_dataset( + client, # type: ignore[arg-type] + resource_id="example_datasets", + dataset_id="EXAMPLE_MEL_RNA", + factor="condition", + group1="MOCK", + group2="MCM", + ) + assert analysis_id == "an-1" + assert client.matrix_downloads == 1 + assert client.submitted["analysis_group"] == ["MCM", "MOCK", "MCM", "MOCK"] + assert client.submitted["group1"] == "MOCK" + + +@asyncio_test +async def test_an_unknown_factor_says_what_there_is() -> None: + client = StubClient(summary=SUMMARY) + with pytest.raises(job.AnalysisFailedError, match="condition"): + await job.submit_public_dataset( + client, # type: ignore[arg-type] + resource_id="example_datasets", + dataset_id="EXAMPLE_MEL_RNA", + factor="treatment", + group1="a", + group2="b", + ) + + +@asyncio_test +async def test_an_unknown_group_says_what_there_is() -> None: + client = StubClient(summary=SUMMARY) + with pytest.raises(job.AnalysisFailedError, match="MCM, MOCK"): + await job.submit_public_dataset( + client, # type: ignore[arg-type] + resource_id="example_datasets", + dataset_id="EXAMPLE_MEL_RNA", + factor="condition", + group1="TREATED", + group2="MOCK", + ) + + +@asyncio_test +async def test_comparing_a_group_with_itself_is_refused() -> None: + client = StubClient(summary=SUMMARY) + with pytest.raises(job.AnalysisFailedError, match="must be different"): + await job.submit_public_dataset( + client, # type: ignore[arg-type] + resource_id="example_datasets", + dataset_id="EXAMPLE_MEL_RNA", + factor="condition", + group1="MCM", + group2="MCM", + ) + + +@asyncio_test +async def test_a_failed_analysis_raises_rather_than_returning_nothing( + tmp_path: Path, +) -> None: + # Measured: 200 on submission, then this. A caller that only checked the + # submission would report success and hand back an empty result. + client = StubClient( + statuses=[AnalysisStatus("failed", "CONNECTION_FORCED - broker closed", 1.0)] + ) + with pytest.raises(job.AnalysisFailedError, match="CONNECTION_FORCED"): + await job.await_result(client, "an-1", out_dir=tmp_path) # type: ignore[arg-type] + + +@asyncio_test +async def test_polling_stops_at_the_deadline(tmp_path: Path) -> None: + # Without a deadline this loops forever against a stuck analysis, and + # the symptom is a chat that never answers. + client = StubClient(statuses=[AnalysisStatus("running", "Permutation 1/1000", 0.1)]) + with pytest.raises(job.AnalysisFailedError, match="did not finish"): + await job.await_result( + client, # type: ignore[arg-type] + "an-1", + out_dir=tmp_path, + deadline_seconds=-1, + ) + + +@asyncio_test +async def test_progress_is_reported_while_running(tmp_path: Path) -> None: + seen: list[str] = [] + + async def record(status: AnalysisStatus) -> None: + seen.append(status.description) + + client = StubClient( + statuses=[ + AnalysisStatus("running", "Permutation 200 / 1000", 0.2), + AnalysisStatus("running", "Permutation 900 / 1000", 0.9), + AnalysisStatus("complete", "Analysis done", 1.0), + ] + ) + await job.await_result( + client, # type: ignore[arg-type] + "an-1", + out_dir=tmp_path, + on_progress=record, + ) + assert "Permutation 200 / 1000" in seen + assert "Analysis done" in seen + + +@asyncio_test +async def test_a_finished_analysis_splits_by_audience(tmp_path: Path) -> None: + finished = await job.await_result( + StubClient(), # type: ignore[arg-type] + "an-1", + out_dir=tmp_path, + ) + + # The file gets every column the service sent. + written = finished.table_path.read_text() + assert "MeanWeightT0" in written + + # The model gets the bounded view and no capability URL. + as_prompt = json.dumps(finished.for_model) + assert "PathwayBrowser" not in as_prompt + assert finished.for_model["no_result"] is False + + # The user gets the link. + assert any("PathwayBrowser" in url for _, url in finished.links) + + +@asyncio_test +async def test_an_uploaded_file_is_deleted_even_when_submission_fails( + tmp_path: Path, +) -> None: + path = tmp_path / "theirs.tsv" + path.write_text("\tS1\tS2\nENSG1\t1\t2\n") + matrix = Matrix( + path=path, size_bytes=path.stat().st_size, samples=["S1", "S2"], gene_count=1 + ) + + class Failing(StubClient): + async def submit(self, **kwargs: Any) -> str: + raise RuntimeError("upstream is down") + + with pytest.raises(RuntimeError): + await job.submit_uploaded_matrix( + Failing(), # type: ignore[arg-type] + matrix=matrix, + dataset_type="rnaseq_counts", + analysis_group=["A", "B"], + group1="A", + group2="B", + ) + assert not path.exists() + + +@asyncio_test +async def test_an_uploaded_file_does_not_carry_its_name_to_the_service( + tmp_path: Path, +) -> None: + # The filename is user free text -- `smith_lab_unpublished_2026.txt` -- + # and it would come back in the result as `datasets[].name`. + path = tmp_path / "smith_lab_unpublished_2026.tsv" + path.write_text("\tS1\tS2\nENSG1\t1\t2\n") + matrix = Matrix(path=path, size_bytes=1, samples=["S1", "S2"], gene_count=1) + + client = StubClient() + await job.submit_uploaded_matrix( + client, # type: ignore[arg-type] + matrix=matrix, + dataset_type="rnaseq_counts", + analysis_group=["A", "B"], + group1="A", + group2="B", + ) + assert "smith_lab" not in json.dumps(client.submitted["dataset_name"]) + + +@asyncio_test +async def test_mismatched_group_labels_are_refused(tmp_path: Path) -> None: + path = tmp_path / "m.tsv" + path.write_text("\tS1\tS2\tS3\nENSG1\t1\t2\t3\n") + matrix = Matrix(path=path, size_bytes=1, samples=["S1", "S2", "S3"], gene_count=1) + + with pytest.raises(job.AnalysisFailedError, match="one label per"): + await job.submit_uploaded_matrix( + StubClient(), # type: ignore[arg-type] + matrix=matrix, + dataset_type="rnaseq_counts", + analysis_group=["A", "B"], + group1="A", + group2="B", + ) + + +@asyncio_test +async def test_polling_is_bounded_by_iterations_as_well_as_time(tmp_path: Path) -> None: + """The wall-clock deadline assumes every turn of the loop waits. + + When sleeping does not sleep -- patched here, but equally a zero + interval from a caller -- elapsed time never advances and the loop + never exits. Discovered by sabotaging the failure check: the suite hung + instead of failing, which is a worse outcome than either. + """ + forever = StubClient( + statuses=[AnalysisStatus("running", "Permutation 1/1000", 0.1)] + ) + + with pytest.raises(job.AnalysisFailedError, match="did not finish"): + await job.await_result( + forever, # type: ignore[arg-type] + "an-1", + out_dir=tmp_path, + deadline_seconds=30 * 60, + poll_interval=0.0, + ) + + +@asyncio_test +async def test_the_upload_is_deleted_when_validation_rejects_it(tmp_path: Path) -> None: + """The leak was on the likeliest path. + + The group checks used to run above the `try`, so a wrong group name -- + one of the two mistakes a user actually makes -- returned an error and + left their matrix on a disk with 4.7 GB free. + """ + path = tmp_path / "theirs.tsv" + path.write_text("\tS1\tS2\nENSG1\t1\t2\n") + matrix = Matrix(path=path, size_bytes=1, samples=["S1", "S2"], gene_count=1) + + with pytest.raises(job.AnalysisFailedError, match="not a value of"): + await job.submit_uploaded_matrix( + StubClient(), # type: ignore[arg-type] + matrix=matrix, + dataset_type="rnaseq_counts", + analysis_group=["A", "B"], + group1="NOPE", + group2="B", + ) + assert not path.exists() + + +@asyncio_test +async def test_the_upload_is_deleted_when_the_label_count_is_wrong( + tmp_path: Path, +) -> None: + path = tmp_path / "theirs.tsv" + path.write_text("\tS1\tS2\tS3\nENSG1\t1\t2\t3\n") + matrix = Matrix(path=path, size_bytes=1, samples=["S1", "S2", "S3"], gene_count=1) + + with pytest.raises(job.AnalysisFailedError, match="one label per"): + await job.submit_uploaded_matrix( + StubClient(), # type: ignore[arg-type] + matrix=matrix, + dataset_type="rnaseq_counts", + analysis_group=["A", "B"], + group1="A", + group2="B", + ) + assert not path.exists() + + +@asyncio_test +async def test_loading_a_dataset_is_bounded_too(tmp_path: Path) -> None: + """The sibling loop. `await_result` got two bounds; this one had one, + twelve lines away, and I missed it in the same review.""" + stuck = StubClient( + loading=[LoadingStatus("running", "still loading", 0.1, None)], + summary=SUMMARY, + ) + with pytest.raises(job.AnalysisFailedError, match="did not finish"): + await job.submit_public_dataset( + stuck, # type: ignore[arg-type] + resource_id="example_datasets", + dataset_id="EXAMPLE_MEL_RNA", + factor="condition", + group1="MOCK", + group2="MCM", + deadline_seconds=30 * 60, + ) + + +def test_an_unknown_terminal_status_counts_as_failure() -> None: + """`failed` is derived from "terminal and not complete". + + If the service adds `cancelled` to its terminal statuses, the derived + form treats it as a failure. An `== "failed"` comparison would have + called it neither finished nor failed and spun until a bound fired, + then reported a timeout for something that had already stopped. + """ + from gsa import client as gsa_client + + cancelled = gsa_client.AnalysisStatus("cancelled", "user cancelled", 1.0) + assert not cancelled.finished # not terminal until the set says so + + with_cancel = gsa_client.TERMINAL_STATUSES | {"cancelled"} + assert "complete" in with_cancel + # The property reads the set, so extending it is the only change needed. + assert gsa_client.AnalysisStatus("complete", "", 1.0).finished + assert not gsa_client.AnalysisStatus("complete", "", 1.0).failed + + +def test_pruning_removes_old_tables_and_keeps_recent_ones(tmp_path: Path) -> None: + import os + + old = tmp_path / "reactome-gsa-old.tsv" + new = tmp_path / "reactome-gsa-new.tsv" + for path in (old, new): + path.write_text("Pathway\tName\n") + two_days = time.time() - 2 * 24 * 60 * 60 + os.utime(old, (two_days, two_days)) + + assert job.prune_results(tmp_path) == 1 + assert not old.exists() + assert new.exists() + + +def test_pruning_bounds_the_total_size(tmp_path: Path) -> None: + import os + + paths = [] + for index in range(5): + path = tmp_path / f"reactome-gsa-{index}.tsv" + path.write_text("x" * 1000) + os.utime(path, (time.time() - (10 - index), time.time() - (10 - index))) + paths.append(path) + + job.prune_results(tmp_path, max_total_bytes=2500) + remaining = sorted(p.name for p in tmp_path.glob("reactome-gsa-*.tsv")) + # Oldest go first, newest survive. + assert remaining == ["reactome-gsa-3.tsv", "reactome-gsa-4.tsv"] + + +def test_pruning_touches_only_files_it_wrote(tmp_path: Path) -> None: + import os + + mine = tmp_path / "reactome-gsa-old.tsv" + theirs = tmp_path / "someone-elses-important.tsv" + for path in (mine, theirs): + path.write_text("data") + two_days = time.time() - 2 * 24 * 60 * 60 + os.utime(mine, (two_days, two_days)) + os.utime(theirs, (two_days, two_days)) + + job.prune_results(tmp_path) + assert not mine.exists() + assert theirs.exists() + + +@asyncio_test +async def test_writing_a_result_prunes_the_directory_first(tmp_path: Path) -> None: + """That `prune_results` works is not the same as it being called. + + The pruning tests all invoked the function directly, so removing the + call from `await_result` broke nothing -- the directory would have + grown forever with every test still green. Found by sabotage: the + deletion the feature promises needs a test on the path that promises + it. + """ + import os + + stale = tmp_path / "reactome-gsa-ancient.tsv" + stale.write_text("Pathway\tName\n") + long_ago = time.time() - 30 * 24 * 60 * 60 + os.utime(stale, (long_ago, long_ago)) + + finished = await job.await_result( + StubClient(), # type: ignore[arg-type] + "an-1", + out_dir=tmp_path, + ) + + assert not stale.exists() + assert finished.table_path.exists() diff --git a/tests/gsa/test_gsa_upload.py b/tests/gsa/test_gsa_upload.py new file mode 100644 index 0000000..bd701e3 --- /dev/null +++ b/tests/gsa/test_gsa_upload.py @@ -0,0 +1,122 @@ +"""Accepting a matrix, and refusing one in terms the user can act on.""" + +from pathlib import Path + +import pytest + +from gsa import upload + + +def write(tmp_path: Path, text: str, name: str = "matrix.tsv") -> Path: + path = tmp_path / name + path.write_text(text) + return path + + +GOOD = "\tS1\tS2\tS3\tS4\nENSG1\t10\t20\t30\t40\nENSG2\t5\t6\t7\t8\n" + + +def test_accepts_a_real_matrix_shape(tmp_path: Path) -> None: + # The header's first cell is empty, which is what the service's own + # export produces -- the measured example began with a tab. + matrix = upload.validate(write(tmp_path, GOOD)) + assert matrix.samples == ["S1", "S2", "S3", "S4"] + assert matrix.gene_count == 2 + + +def test_accepts_comma_separated(tmp_path: Path) -> None: + matrix = upload.validate(write(tmp_path, GOOD.replace("\t", ","), "m.csv")) + assert matrix.samples == ["S1", "S2", "S3", "S4"] + + +def test_a_comma_inside_a_tsv_does_not_split_the_header(tmp_path: Path) -> None: + # Tab wins when both are present. Splitting on the comma would turn + # "Tumour, left" into two samples and mangle the analysis rather than + # fail it. + text = "\tTumour, left\tTumour, right\nENSG1\t1\t2\n" + assert upload.validate(write(tmp_path, text)).samples == [ + "Tumour, left", + "Tumour, right", + ] + + +def test_refuses_a_file_over_the_cap( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv(upload.MAX_UPLOAD_BYTES_ENV, "100") + with pytest.raises(upload.UploadRejectedError, match="limit"): + upload.validate(write(tmp_path, GOOD * 100)) + + +def test_the_cap_defaults_well_below_chainlit(tmp_path: Path) -> None: + # Chainlit ships max_size_mb = 500 and the host has 4.7 GB free. + assert upload.max_upload_bytes() == 20 * 1024 * 1024 + + +@pytest.mark.parametrize("bad", ["", "0", "-1", "not-a-number"]) +def test_an_unusable_cap_falls_back_rather_than_disabling_the_limit( + bad: str, monkeypatch: pytest.MonkeyPatch +) -> None: + # The dangerous direction: a typo must not mean "no limit". + monkeypatch.setenv(upload.MAX_UPLOAD_BYTES_ENV, bad) + assert upload.max_upload_bytes() == upload.DEFAULT_MAX_UPLOAD_BYTES + + +def test_refuses_an_empty_file(tmp_path: Path) -> None: + with pytest.raises(upload.UploadRejectedError, match="empty"): + upload.validate(write(tmp_path, "")) + + +def test_refuses_something_that_is_not_a_matrix(tmp_path: Path) -> None: + with pytest.raises(upload.UploadRejectedError, match="expression matrix"): + upload.validate(write(tmp_path, "just some prose\nand another line\n")) + + +def test_refuses_a_header_with_no_data(tmp_path: Path) -> None: + with pytest.raises(upload.UploadRejectedError, match="no data rows"): + upload.validate(write(tmp_path, "\tS1\tS2\tS3\n")) + + +def test_refuses_a_single_sample(tmp_path: Path) -> None: + # An analysis compares two groups. One column cannot be compared with + # anything, and the service would say so minutes later in R's words. + text = "gene\tS1\tS2\nENSG1\t1\t2\n" + matrix = upload.validate(write(tmp_path, text)) + assert len(matrix.samples) == 2 + + with pytest.raises(upload.UploadRejectedError, match="only one sample"): + upload.validate(write(tmp_path, "gene\tS1\nENSG1\t1\nENSG2\t2\n", "one.tsv")) + + +def test_discard_removes_the_file(tmp_path: Path) -> None: + path = write(tmp_path, GOOD) + upload.discard(path) + assert not path.exists() + + +def test_discard_is_safe_to_repeat(tmp_path: Path) -> None: + # Called from a `finally`, so it runs on paths that may already be gone. + path = write(tmp_path, GOOD) + upload.discard(path) + upload.discard(path) + + +def test_a_large_file_reports_an_unknown_gene_count_not_a_sentinel( + tmp_path: Path, +) -> None: + """Counting every row of a 20,000-gene file answers a question nobody + asked, so it stops early -- but "stopped early" must not be encoded as + a number. A `-1` here is the kind of value that reaches a user as + "your matrix has -1 genes". + """ + rows = "".join(f"G{i}\t1\t2\n" for i in range(6000)) + matrix = upload.validate(write(tmp_path, "\tS1\tS2\n" + rows, "big.tsv")) + + assert matrix.gene_count is None + assert matrix.samples == ["S1", "S2"] + + +def test_a_small_file_still_reports_a_real_count(tmp_path: Path) -> None: + # The control: if every file reported None, the test above would pass + # and the count would be useless. + assert upload.validate(write(tmp_path, GOOD)).gene_count == 2