diff --git a/.github/workflows/server-test.yml b/.github/workflows/server-test.yml index 817f2ce..9eabf3b 100644 --- a/.github/workflows/server-test.yml +++ b/.github/workflows/server-test.yml @@ -36,6 +36,9 @@ jobs: cache: pip cache-dependency-path: server/requirements-dev.txt + - name: Install system dependencies + run: sudo apt-get update && sudo apt-get install -y --no-install-recommends tabix + - name: Install dependencies run: pip install -r requirements-dev.txt diff --git a/server/jobs/repair_annotation_files.py b/server/jobs/repair_annotation_files.py index 2b8b5c4..f00e8a6 100644 --- a/server/jobs/repair_annotation_files.py +++ b/server/jobs/repair_annotation_files.py @@ -19,6 +19,7 @@ - Run with dry_run=True first (the default) to review what would be repaired before actually downloading/writing anything. """ +import gzip import os import shlex import shutil @@ -140,7 +141,9 @@ def recreate_annotation_file(annotation: GenomeAnnotation, tmp_dir: str, downloa tmp_subdir_path = file_helper.create_dir_path(tmp_dir, f"repair_{annotation.annotation_id}") try: - gzipped_downloaded_path = os.path.join(tmp_subdir_path, "source_download") + # Must end in .gz so _gff_decompress_cmd selects zcat (not cat). All tracker + # source URLs are gzip-compressed, matching the main import job's assumption. + gzipped_downloaded_path = os.path.join(tmp_subdir_path, "source_download.gz") with requests.get(download_url, stream=True, timeout=DOWNLOAD_TIMEOUT) as r: r.raise_for_status() with open(gzipped_downloaded_path, "wb") as f: @@ -150,13 +153,24 @@ def recreate_annotation_file(annotation: GenomeAnnotation, tmp_dir: str, downloa if file_helper.file_is_empty_or_does_not_exist(gzipped_downloaded_path): raise Exception(f"Downloaded content from {download_url} is empty") + # Reject truncated/non-gzip payloads before the shell pipeline; otherwise a + # silent empty stream can look like a ContentMismatchError. + try: + with gzip.open(gzipped_downloaded_path, "rb") as gz_f: + if not gz_f.read(1): + raise Exception(f"Downloaded gzip from {download_url} decompresses to empty content") + except OSError as e: + raise Exception(f"Downloaded content from {download_url} is not valid gzip: {e}") from e + tmp_bgzipped_path = os.path.join(tmp_subdir_path, "output.gff.gz") md5_path = os.path.join(tmp_subdir_path, "md5.txt") # Reuse the exact same sort/tee/bgzip shell pipeline as the main import job # (jobs/services/annotation.py::process_annotation_file) so the recreated - # content is produced identically. + # content is produced identically. pipefail so a failed zcat/grep/sort stage + # cannot be masked by bgzip succeeding on empty stdin. stream_cmd = ( + "set -o pipefail; " f"{annotation_service._sorted_gff_stream_cmd(gzipped_downloaded_path)} " f"| tee >(md5sum | awk '{{print $1}}' > {shlex.quote(md5_path)}) " f"| bgzip > {shlex.quote(tmp_bgzipped_path)}" diff --git a/server/tests/unit/jobs/test_repair_annotation_files.py b/server/tests/unit/jobs/test_repair_annotation_files.py index 4427bd7..6ba0954 100644 --- a/server/tests/unit/jobs/test_repair_annotation_files.py +++ b/server/tests/unit/jobs/test_repair_annotation_files.py @@ -1,12 +1,43 @@ +import gzip +import hashlib import os +import shutil +import subprocess +import tempfile from unittest.mock import MagicMock, patch import pytest from jobs import repair_annotation_files as repair_job +from jobs.services import annotation as annotation_service pytestmark = pytest.mark.unit +# Minimal unsorted GFF3 used by the real (non-mocked) pipeline regression test. +_REAL_PIPELINE_GFF = ( + "##gff-version 3\n" + "##sequence-region chr2 1 100\n" + "chr2\tsource\tgene\t10\t20\t.\t+\t.\tID=g2\n" + "chr1\tsource\tgene\t5\t15\t.\t+\t.\tID=g1\n" + "chr1\tsource\texon\t5\t10\t.\t+\t.\tID=e1;Parent=g1\n" +) + +_HAS_BGZIP_TABIX = shutil.which("bgzip") is not None and shutil.which("tabix") is not None + + +def _sorted_gff_md5_via_shell(gff_text: str) -> str: + """ + MD5 of the sorted uncompressed stream using the same shell command the + repair/import jobs use (_sorted_gff_stream_cmd). Requires zcat/grep/sort only. + """ + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "fixture.gff.gz") + with gzip.open(path, "wb") as f: + f.write(gff_text.encode()) + cmd = annotation_service._sorted_gff_stream_cmd(path) + proc = subprocess.run(["bash", "-lc", cmd], capture_output=True, check=True) + return hashlib.md5(proc.stdout).hexdigest() + def _fake_popen_factory(tmp_subdir_path, bgzipped_path, content_md5): """ @@ -40,7 +71,11 @@ def communicate(): return fake_popen -def _fake_download(content=b"not-empty-source-content"): +def _fake_download(content=None): + """Default content is a tiny valid gzip payload (recreate validates gzip).""" + if content is None: + content = gzip.compress(b"##gff-version 3\n") + def _get(url, stream=True, timeout=None): cm = MagicMock() cm.__enter__.return_value = cm @@ -174,6 +209,113 @@ def test_content_mismatch_does_not_touch_final_path(self, tmp_path): # temp working dir cleaned up even on failure assert not os.path.exists(tmp_subdir_path) + def test_rejects_non_gzip_download_before_pipeline(self, tmp_path): + """Invalid gzip must raise a plain Exception, not ContentMismatchError.""" + final_bgzipped_path = str(tmp_path / "final" / "Ensembl_x.gff.gz") + tmp_root = str(tmp_path / "tmp") + + annotation = MagicMock() + annotation.annotation_id = "expectedmd5expectedmd5expected1" + + with ( + patch.object(repair_job.file_helper, "get_annotation_file_path", return_value=final_bgzipped_path), + patch.object( + repair_job.requests, + "get", + side_effect=_fake_download(content=b"this-is-not-gzip"), + ), + patch.object(repair_job.subprocess, "Popen") as popen, + ): + with pytest.raises(Exception, match="not valid gzip"): + repair_job.recreate_annotation_file( + annotation, tmp_root, "http://example.com/a.gff.gz" + ) + + popen.assert_not_called() + assert not os.path.exists(final_bgzipped_path) + + def test_stream_cmd_uses_gz_suffix_zcat_and_pipefail(self, tmp_path): + """ + Regression for the production bug where the download temp file had no + .gz suffix, so _gff_decompress_cmd selected cat instead of zcat and the + pipeline silently hashed empty content. + """ + content_md5 = "deadbeefdeadbeefdeadbeefdeadbeef" + final_bgzipped_path = str(tmp_path / "final" / "Ensembl_x.gff.gz") + tmp_root = str(tmp_path / "tmp") + tmp_subdir_path = os.path.join(tmp_root, f"repair_{content_md5}") + tmp_bgzipped_path = os.path.join(tmp_subdir_path, "output.gff.gz") + fake_popen = _fake_popen_factory(tmp_subdir_path, tmp_bgzipped_path, content_md5) + + annotation = MagicMock() + annotation.annotation_id = content_md5 + captured_cmds = [] + + def capturing_popen(cmd, stdout=None, stderr=None): + captured_cmds.append(cmd) + return fake_popen(cmd, stdout=stdout, stderr=stderr) + + with ( + patch.object(repair_job.file_helper, "get_annotation_file_path", return_value=final_bgzipped_path), + patch.object(repair_job.requests, "get", side_effect=_fake_download()), + patch.object(repair_job, "seq_files"), + patch.object(repair_job.subprocess, "Popen", side_effect=capturing_popen), + ): + repair_job.recreate_annotation_file( + annotation, tmp_root, "http://example.com/a.gff.gz" + ) + + assert captured_cmds, "expected at least the stream Popen call" + stream_argv = captured_cmds[0] + assert stream_argv[:2] == ["bash", "-lc"] + stream_cmd = stream_argv[2] + assert "set -o pipefail" in stream_cmd + assert "source_download.gz" in stream_cmd + assert "zcat" in stream_cmd + # Must not use the unsuffixed path that previously selected `cat`. + assert "/source_download " not in stream_cmd + assert "/source_download|" not in stream_cmd + assert "/source_download'" not in stream_cmd + assert '/source_download"' not in stream_cmd + + @pytest.mark.skipif(not _HAS_BGZIP_TABIX, reason="bgzip/tabix not installed") + def test_real_pipeline_decompresses_gzip_and_matches_sorted_md5(self, tmp_path): + """ + End-to-end regression for the missing-.gz-suffix bug: leave subprocess + unmocked so zcat|sort|bgzip|tabix actually run. Would fail with + ContentMismatchError (empty-content md5) if the download temp path did + not end in .gz. + """ + expected_md5 = _sorted_gff_md5_via_shell(_REAL_PIPELINE_GFF) + gzipped_source = gzip.compress(_REAL_PIPELINE_GFF.encode()) + final_bgzipped_path = str(tmp_path / "final" / "Ensembl_x.gff.gz") + tmp_root = str(tmp_path / "tmp") + + annotation = MagicMock() + annotation.annotation_id = expected_md5 + + with ( + patch.object(repair_job.file_helper, "get_annotation_file_path", return_value=final_bgzipped_path), + patch.object( + repair_job.requests, + "get", + side_effect=_fake_download(content=gzipped_source), + ), + patch.object(repair_job, "seq_files") as seq_files_mock, + ): + md5_result, file_size = repair_job.recreate_annotation_file( + annotation, tmp_root, "http://example.com/a.gff.gz" + ) + + assert md5_result == expected_md5 + assert file_size > 0 + assert os.path.exists(final_bgzipped_path) + assert os.path.getsize(final_bgzipped_path) > 0 + assert os.path.exists(f"{final_bgzipped_path}.csi") + assert os.path.getsize(f"{final_bgzipped_path}.csi") > 0 + seq_files_mock.write_contigs_from_tabix.assert_called_once_with(final_bgzipped_path) + assert not os.path.exists(os.path.join(tmp_root, f"repair_{expected_md5}")) + class TestRepairMissingAnnotationFiles: def test_dry_run_reports_without_repairing(self):