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