|
| 1 | +"""Exact original-confidence inventory through the existing download owner.""" |
| 2 | +import hashlib |
| 3 | +import json |
| 4 | +import os |
| 5 | +from pathlib import Path |
| 6 | + |
| 7 | +from fastapi import FastAPI |
| 8 | +from httpx import ASGITransport, AsyncClient |
| 9 | +import pytest |
| 10 | +from sqlalchemy import create_engine, select |
| 11 | +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine |
| 12 | + |
| 13 | +from database import Base, Design, Job, get_session |
| 14 | +from routers import designs, files |
| 15 | +from services import protenix_scientific_consumer as consumer |
| 16 | +from test_protenix_native_confidence import make_publication, store |
| 17 | + |
| 18 | + |
| 19 | +def app_for(session): |
| 20 | + app = FastAPI() |
| 21 | + app.include_router(designs.router, prefix='/api/designs') |
| 22 | + app.include_router(files.router, prefix='/api/files') |
| 23 | + async def dependency(): |
| 24 | + yield session |
| 25 | + app.dependency_overrides[get_session] = dependency |
| 26 | + app.dependency_overrides[files.get_governed_ngs_result_roots] = lambda: () |
| 27 | + return app |
| 28 | + |
| 29 | + |
| 30 | +async def check_inventory(session, design_id): |
| 31 | + row = await session.get(Design, design_id) |
| 32 | + selected = await consumer.verified_native_design(row, session) |
| 33 | + async with AsyncClient(transport=ASGITransport(app=app_for(session)), base_url='http://test') as client: |
| 34 | + response = await client.get(f'/api/designs/{design_id}/confidence-artifacts') |
| 35 | + assert response.status_code == 200, response.text |
| 36 | + payload = response.json() |
| 37 | + assert payload['status'] == 'ok', payload |
| 38 | + assert payload['producer'] == 'protenix' |
| 39 | + assert {a['role'] for a in payload['artifacts']} == {'structure', 'metrics', 'pae'} |
| 40 | + for artifact in payload['artifacts']: |
| 41 | + role = artifact['role'] |
| 42 | + expected = selected['artifacts'][role] |
| 43 | + assert artifact['retained'] and artifact['filename'] == Path(expected['path']).name |
| 44 | + download = await client.get(artifact['download_url']) |
| 45 | + assert download.status_code == 200, download.text |
| 46 | + assert download.content == selected['snapshots'][role] |
| 47 | + assert hashlib.sha256(download.content).hexdigest() == artifact['artifact_sha256'] == expected['sha256'] |
| 48 | + assert len(download.content) == artifact['size_bytes'] |
| 49 | + assert not session.dirty and not session.new |
| 50 | + return payload |
| 51 | + |
| 52 | + |
| 53 | +@pytest.mark.asyncio |
| 54 | +@pytest.mark.parametrize('custody', ['local', 'remote', 'retained_remote']) |
| 55 | +@pytest.mark.parametrize('fault', [None, 'missing', 'changed', 'foreign']) |
| 56 | +async def test_exact_confidence_inventory(tmp_path, monkeypatch, custody, fault): |
| 57 | + import paths |
| 58 | + job, design, root = make_publication(tmp_path, custody=custody) |
| 59 | + monkeypatch.setattr(paths, 'get_allowed_roots', lambda: {'bms_results': root}) |
| 60 | + engine, factory = await store(tmp_path, job, design) |
| 61 | + try: |
| 62 | + async with factory() as session: |
| 63 | + row = await session.get(Design, design.id) |
| 64 | + selected = await consumer.verified_native_design(row, session) |
| 65 | + full = Path(selected['artifacts']['pae']['path']) |
| 66 | + # A different input with the same sample suffix is never a fallback. |
| 67 | + full.with_name('foreign_full_data_sample_0.json').write_bytes(full.read_bytes()) |
| 68 | + if fault == 'missing': |
| 69 | + full.unlink() |
| 70 | + elif fault == 'changed': |
| 71 | + full.write_bytes(b'{}') |
| 72 | + elif fault == 'foreign': |
| 73 | + row.json_path = str(full.with_name('foreign_summary_confidence_sample_0.json')) |
| 74 | + await session.commit() |
| 75 | + if fault is None: |
| 76 | + await check_inventory(session, design.id) |
| 77 | + else: |
| 78 | + async with AsyncClient(transport=ASGITransport(app=app_for(session)), base_url='http://test') as client: |
| 79 | + response = await client.get(f'/api/designs/{design.id}/confidence-artifacts') |
| 80 | + assert response.status_code == 200 |
| 81 | + assert response.json()['status'] == 'unavailable' |
| 82 | + assert response.json()['artifacts'] == [] |
| 83 | + # Optional inventory failure does not remove the structure. |
| 84 | + structure = await client.get(f'/api/designs/{design.id}/pdb') |
| 85 | + assert structure.status_code == 200 |
| 86 | + finally: |
| 87 | + await engine.dispose() |
| 88 | + |
| 89 | + |
| 90 | +@pytest.mark.asyncio |
| 91 | +async def test_retained_inventory_readonly(tmp_path, monkeypatch): |
| 92 | + """Opt-in real rows copied with Core; retained native files only read.""" |
| 93 | + database = os.environ.get('BMS_PROTENIX_INVENTORY_DATABASE') |
| 94 | + if not database: |
| 95 | + pytest.skip('read-only retained Protenix database not selected') |
| 96 | + import paths |
| 97 | + ids = json.loads(os.environ['BMS_PROTENIX_INVENTORY_DESIGN_IDS']) |
| 98 | + assert isinstance(ids, list) and ids and all(isinstance(value, str) for value in ids) |
| 99 | + source = create_engine(f'sqlite:///file:{database}?mode=ro&uri=true') |
| 100 | + # Copy only exact referenced Parquet files before typed JSON resolution. |
| 101 | + import shutil |
| 102 | + from sqlalchemy import text |
| 103 | + copied_root = tmp_path / 'scientific_artifacts' |
| 104 | + monkeypatch.setenv('BMS_SCIENTIFIC_ARTIFACT_ROOT', str(copied_root)) |
| 105 | + def copy_references(value): |
| 106 | + if isinstance(value, str): |
| 107 | + try: |
| 108 | + value = json.loads(value) |
| 109 | + except (ValueError, TypeError): |
| 110 | + return |
| 111 | + if isinstance(value, dict): |
| 112 | + if value.get('schema') == 'bms.scientific-artifact-row-reference.v1': |
| 113 | + from services.scientific_artifacts.contracts import require_row_reference |
| 114 | + relative = Path(require_row_reference(value)['relative_path']) |
| 115 | + assert not relative.is_absolute() and '..' not in relative.parts |
| 116 | + destination = copied_root / relative |
| 117 | + destination.parent.mkdir(parents=True, exist_ok=True) |
| 118 | + shutil.copyfile(Path(database).parent / 'scientific_artifacts' / relative, destination) |
| 119 | + for item in value.values(): |
| 120 | + copy_references(item) |
| 121 | + elif isinstance(value, list): |
| 122 | + for item in value: |
| 123 | + copy_references(item) |
| 124 | + with source.connect() as conn: |
| 125 | + for design_id in ids: |
| 126 | + raw = conn.execute(text('SELECT * FROM designs WHERE id = :id'), {'id': design_id}).mappings().one() |
| 127 | + for value in raw.values(): |
| 128 | + copy_references(value) |
| 129 | + design_rows = [dict(r) for r in conn.execute(select(Design.__table__).where(Design.id.in_(ids))).mappings()] |
| 130 | + jobs = {row['job_id'] for row in design_rows} |
| 131 | + job_rows = [dict(r) for r in conn.execute(select(Job.__table__).where(Job.id.in_(jobs))).mappings()] |
| 132 | + source.dispose() |
| 133 | + assert len(job_rows) == len(jobs) and len(design_rows) == len(ids) |
| 134 | + monkeypatch.setattr(paths, 'get_allowed_roots', lambda: {'bms_results': Path(database).parent / 'bms_results'}) |
| 135 | + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "retained-copy.sqlite"}') |
| 136 | + async with engine.begin() as conn: |
| 137 | + await conn.run_sync(Base.metadata.create_all) |
| 138 | + # No ORM flush/publication hooks and no live-store writes. |
| 139 | + await conn.execute(Job.__table__.insert(), job_rows) |
| 140 | + await conn.execute(Design.__table__.insert(), design_rows) |
| 141 | + factory = async_sessionmaker(engine, expire_on_commit=False) |
| 142 | + try: |
| 143 | + async with factory() as session: |
| 144 | + reports = [await check_inventory(session, design_id) for design_id in ids] |
| 145 | + output = os.environ.get('BMS_PROTENIX_INVENTORY_REPORT') |
| 146 | + if output: |
| 147 | + Path(output).write_text(json.dumps(reports, indent=2)) |
| 148 | + finally: |
| 149 | + await engine.dispose() |
0 commit comments