|
| 1 | +"""HTTP transport fixtures, not RFD3 execution/publication acceptance.""" |
| 2 | +import gzip |
| 3 | +import hashlib |
| 4 | +from types import SimpleNamespace |
| 5 | +from unittest.mock import AsyncMock |
| 6 | + |
| 7 | +import httpx |
| 8 | +import pytest |
| 9 | +from fastapi import FastAPI |
| 10 | +from routers import jobs |
| 11 | + |
| 12 | + |
| 13 | +@pytest.mark.asyncio |
| 14 | +@pytest.mark.parametrize("owner", ["generation", "local_redesign"]) |
| 15 | +@pytest.mark.parametrize("compressed", [False, True]) |
| 16 | +async def test_rfd3_download_filename_matches_http_decoded_content(tmp_path, owner, compressed): |
| 17 | + # The database read is an inert transport double. Integrity/path checks and |
| 18 | + # actual FileResponse streaming remain the production implementation. |
| 19 | + content = b"data_transport_fixture\n#\n" if compressed else b'{"transport_fixture":true}\n' |
| 20 | + filename = "candidate.cif.gz" if compressed else "metadata.json" |
| 21 | + native_bytes = gzip.compress(content, mtime=0) if compressed else content |
| 22 | + path = tmp_path / filename |
| 23 | + path.write_bytes(native_bytes) |
| 24 | + digest = hashlib.sha256(native_bytes).hexdigest() |
| 25 | + media_type = "chemical/x-mmcif+gzip" if compressed else "application/json" |
| 26 | + role = "candidate_structure" if compressed else "candidate_metadata" |
| 27 | + job = SimpleNamespace(id="transport-job", model_id="protein_local_redesign", output_dir=str(tmp_path)) |
| 28 | + if owner == "generation": |
| 29 | + item = SimpleNamespace(provenance={"artifacts": [{ |
| 30 | + "role": role, "relative_path": filename, "bytes": len(native_bytes), |
| 31 | + "sha256": digest, "media_type": media_type, |
| 32 | + }]}) |
| 33 | + else: |
| 34 | + item = SimpleNamespace(role="structure" if compressed else "native_prediction_metadata", |
| 35 | + relative_path=filename, storage_path=str(path), |
| 36 | + size_bytes=len(native_bytes), content_sha256=digest, |
| 37 | + media_type=media_type) |
| 38 | + session = SimpleNamespace(execute=AsyncMock(side_effect=[ |
| 39 | + SimpleNamespace(scalar_one_or_none=lambda: job), |
| 40 | + SimpleNamespace(scalar_one_or_none=lambda: item), |
| 41 | + ])) |
| 42 | + app = FastAPI() |
| 43 | + |
| 44 | + @app.get("/artifact") |
| 45 | + async def artifact(): |
| 46 | + if owner == "generation": |
| 47 | + return await jobs.get_rfd3_generation_artifact(job.id, "candidate", role, session) |
| 48 | + return await jobs.get_rfd3_local_redesign_artifact(job.id, "artifact", session) |
| 49 | + |
| 50 | + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: |
| 51 | + response = await client.get("/artifact") |
| 52 | + assert response.status_code == 200, response.text |
| 53 | + assert response.content == content # HTTPX and browsers decode the gzip transport. |
| 54 | + expected_name = "candidate.cif" if compressed else "metadata.json" |
| 55 | + assert response.headers["content-disposition"] == f'attachment; filename="{expected_name}"' |
| 56 | + assert response.headers.get("content-encoding") == ("gzip" if compressed else None) |
| 57 | + assert path.read_bytes() == native_bytes |
| 58 | + assert hashlib.sha256(path.read_bytes()).hexdigest() == digest |
0 commit comments