78 lines
2.6 KiB
Python
Executable File
78 lines
2.6 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import zipfile
|
|
|
|
|
|
def test_reference_bundle_download_reports_progress_events(tmp_path, monkeypatch):
|
|
from apps.web import pipeline_backend as backend
|
|
|
|
job = backend.JobInfo(
|
|
job_id="job_test_ref_bundle_progress",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
(workspace / "final").mkdir(parents=True)
|
|
(workspace / "logs").mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
|
|
fake_zip = tmp_path / "ncbi_dataset.zip"
|
|
with zipfile.ZipFile(fake_zip, "w") as zf:
|
|
zf.writestr("ncbi_dataset/data/GCF_FAKE/protein.faa", ">SOE_PEP1\nMAAA\n")
|
|
zf.writestr(
|
|
"ncbi_dataset/data/GCF_FAKE/cds_from_genomic.fna",
|
|
">lcl|NC_000001.1_cds_XP_000001.1_1 [protein_id=XP_000001.1]\nATGGCG\n",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
backend,
|
|
"_select_ncbi_reference_assembly",
|
|
lambda **kwargs: {
|
|
"accession": "GCF_FAKE",
|
|
"organism_name": "Spinacia oleracea",
|
|
"taxon_id": "3562",
|
|
"assembly_name": "fake",
|
|
"assembly_level": "Chromosome",
|
|
"annotation_name": "fake annotation",
|
|
},
|
|
)
|
|
|
|
def fake_download(_url, output_path, timeout_sec=300, progress_callback=None):
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = fake_zip.read_bytes()
|
|
if progress_callback:
|
|
progress_callback({
|
|
"phase": "download",
|
|
"bytes_downloaded": len(data),
|
|
"total_bytes": len(data),
|
|
"percentage": 100,
|
|
})
|
|
output_path.write_bytes(data)
|
|
return output_path.stat().st_size
|
|
|
|
monkeypatch.setattr(backend, "_download_url_to_file", fake_download)
|
|
events = []
|
|
|
|
manifest = backend.download_reference_annotation_bundle(
|
|
job=job,
|
|
species_code="soe",
|
|
species_name="Spinacia oleracea",
|
|
taxon_id="3562",
|
|
requested_stages=["ppi"],
|
|
progress_callback=events.append,
|
|
)
|
|
|
|
phases = [event.get("phase") for event in events]
|
|
assert manifest["status"] == "success"
|
|
assert "select_assembly" in phases
|
|
assert "download" in phases
|
|
assert "extract" in phases
|
|
assert "validate_ids" in phases
|
|
assert phases[-1] == "completed"
|
|
assert manifest["requested_stages"] == ["ppi"]
|
|
assert manifest["requested_file_types"] == ["pep"]
|
|
assert "cds" not in manifest["files"]
|