import json from dataclasses import replace import pytest pytest.importorskip('starlette') pytest.importorskip('httpx') pytest.importorskip('multipart') from starlette.testclient import TestClient from viano_bug_analysis.api import APIConfig, Client, Project, IntakeLimits from viano_bug_analysis.api.app import create_app from viano_bug_analysis.api.config import APIError from viano_bug_analysis.lifecycle.checkpoints import CheckpointStore from viano_bug_analysis.lifecycle.storage import Workspace from viano_bug_analysis.package import PackageReader TOKEN = 'test_token_' + 'x' * 40 @pytest.fixture def api(tmp_path, monkeypatch): monkeypatch.setenv('TEST_INTAKE_TOKEN', TOKEN) config = APIConfig(tmp_path / 'service', (Project('core', 'wow', 'vcore'),), (Client('capture', 'TEST_INTAKE_TOKEN', ('core',)),), limits=IntakeLimits(min_free_bytes=1)) app = create_app(config) return TestClient(app), app.state.intake, config def data(kind='new_case'): result = {'kind': kind, 'project_id': 'core', 'profile_id': 'vcore', 'case_id': 'BUG-000020', 'report': {'observations': ['The observed bug persists.']}} if kind == 'case_return': result['feedback'] = {'outcome': 'patch_failed', 'summary': 'Still broken', 'observations': ['The tested patch did not fix it.']} return result def post(client, source, *, metadata=None, key='first', role='wowdiag', token=TOKEN): return client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + token, 'Idempotency-Key': key}, data={'metadata': json.dumps(metadata or data())}, files={role: ('../untrusted.bin', source.read_bytes(), 'application/octet-stream')}) def test_create_replay_conflict_and_private_status(api, source_pair): client, service, _ = api source = source_pair[0] response = post(client, source) assert response.status_code == 201, response.text receipt = response.json() assert receipt['status'] == 'QUEUED' assert post(client, source).status_code == 200 assert len(list(service.workspace.inbox.glob('*.zip'))) == 1 changed = data() | {'report': {'title': 'Changed'}} assert post(client, source, metadata=changed).status_code == 409 assert post(client, source, key='different').status_code == 409 assert post(client, source, token='bad').status_code == 401 assert client.get('/v1/submissions/' + receipt['submission_id']).status_code == 401 with PackageReader(next(service.workspace.inbox.glob('*.zip'))) as reader: assert reader.manifest['project']['id'] == 'core' assert reader.manifest['state'] == 'NEW' assert b'The observed bug persists.' in reader.read_bytes( 'intake/submissions/' + receipt['submission_id'] + '/report.json') def test_return_preserves_history_and_increments_revision_once(api, source_pair, tmp_path): client, service, _ = api first = post(client, source_pair[0]).json() inbox = next(service.workspace.inbox.glob('*.zip')) # Standalone terminal case simulates transfer back from a different deployment. from viano_bug_analysis.package import PackageWriter terminal = tmp_path / 'terminal.zip' PackageWriter().write(inbox, terminal, manifest_updates={'state': 'OUTPUT_READY', 'analysis_revision': 1, 'history': {'analysis_revisions': 1}, 'routing': {'destination': 'outbox'}}) inbox.unlink() response = post(client, terminal, metadata=data('case_return'), key='return', role='case_package') assert response.status_code == 201, response.text returned = next(service.workspace.inbox.glob('*.zip')) with PackageReader(terminal) as old, PackageReader(returned) as new: assert new.manifest['state'] == 'REOPENED' assert new.manifest['analysis_revision'] == 2 assert new.manifest['history']['feedback'] == 1 assert new.manifest['package_id'] == first['package_id'] for ref in old.files: if ref != 'manifest.json': assert old.read_bytes(ref) == new.read_bytes(ref) index = json.loads(new.read_bytes(new.manifest['artifacts']['case_inputs'])) assert any(b'The tested patch did not fix it.' in new.read_bytes(ref) for ref in index['reports']) assert post(client, terminal, metadata=data('case_return'), key='return', role='case_package').status_code == 200 assert post(client, terminal, metadata=data('case_return'), key='return2', role='case_package').status_code == 409 from viano_bug_analysis.case import CaseBuilder isolated = CheckpointStore(Workspace(tmp_path / 'analysis')) isolated.intake(returned) built = CaseBuilder(isolated).build('BUG-000020', first['package_id']) with PackageReader(built.package) as reader: case = json.loads(reader.read_bytes(built.case_ref)) assert any(o['text'] == 'The tested patch did not fix it.' for o in case['human_observations']['observations']) def test_active_case_cannot_be_returned(api, source_pair): client, service, _ = api post(client, source_pair[0]) source = next(service.workspace.inbox.glob('*.zip')) assert post(client, source, metadata=data('case_return'), key='return', role='case_package').status_code == 409 def test_limits_and_contract(api, source_pair): client, _, config = api assert client.get('/healthz').json() == {'status': 'OK'} assert client.get('/v1/schema', headers={'Authorization': 'Bearer ' + TOKEN}).status_code == 200 assert post(client, source_pair[0], metadata=data() | {'budget': 99}).status_code == 422 assert post(client, source_pair[0], metadata=data() | {'project_id': 'elsewhere'}).status_code == 403 tiny = replace(config, workspace=config.workspace.parent / 'tiny', limits=replace(config.limits, max_body_bytes=100)) limited = TestClient(create_app(tiny)) assert post(limited, source_pair[0]).status_code == 413 def test_recover_prepared_after_publication_interruption(api, source_pair, monkeypatch): client, service, _ = api publish = service._publish def interrupted(receipt): raise OSError('simulated crash') monkeypatch.setattr(service, '_publish', interrupted) with pytest.raises(OSError): service.submit(service.config.clients[0], 'crash', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) monkeypatch.setattr(service, '_publish', publish) service.recover() assert len(list(service.workspace.inbox.glob('*.zip'))) == 1 result, replay = service.submit(service.config.clients[0], 'crash', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) assert replay and result['status'] == 'QUEUED' def test_published_receipt_crash_replay_after_worker_consumes(api, source_pair, monkeypatch): import viano_bug_analysis.api.service as module client, service, _ = api original = module.atomic_json def interrupted(path, value): if value.get('status') == 'QUEUED': raise OSError('crash after publication') original(path, value) monkeypatch.setattr(module, 'atomic_json', interrupted) with pytest.raises(OSError): service.submit(service.config.clients[0], 'crash2', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) inbox = next(service.workspace.inbox.glob('*.zip')) service.store.intake(inbox) inbox.unlink() monkeypatch.setattr(module, 'atomic_json', original) service.recover() assert not list(service.workspace.inbox.glob('*.zip')) result, replay = service.submit(service.config.clients[0], 'crash2', json.dumps(data()).encode(), [('wowdiag', source_pair[0])]) assert replay and result['status'] == 'VALIDATED' def test_file_limit_and_partial_upload_cleanup(api, source_pair): _, _, config = api tiny = replace(config, workspace=config.workspace.parent / 'files', limits=replace(config.limits, max_file_bytes=5)) app = create_app(tiny) response = post(TestClient(app), source_pair[0]) assert response.status_code == 413 assert not list(app.state.intake.staging.iterdir()) def test_return_with_stale_managed_head_is_rejected(api, source_pair, tmp_path): client, service, _ = api first = post(client, source_pair[0]).json() inbox = next(service.workspace.inbox.glob('*.zip')) from viano_bug_analysis.package import PackageWriter terminal = tmp_path / 'terminal.zip' PackageWriter().write(inbox, terminal, manifest_updates={'state': 'OUTPUT_READY', 'analysis_revision': 1, 'history': {'analysis_revisions': 1}, 'routing': {'destination': 'outbox'}}) service.store.intake(terminal) inbox.unlink() response = post(client, terminal, metadata=data('case_return'), key='stale', role='case_package') assert response.status_code == 409 head = service.store.latest('BUG-000020', first['package_id']) response = post(client, head, metadata=data('case_return'), key='fresh', role='case_package') assert response.status_code == 201, response.text assert response.json()['status'] == 'QUEUED' # not the previous OUTPUT_READY result auth = {'Authorization': 'Bearer ' + TOKEN} assert client.get('/v1/submissions/' + response.json()['submission_id'] + '/package', headers=auth).status_code == 409 def test_scope_invalid_json_and_mismatched_source(api, source_pair): client, service, _ = api response = post(client, source_pair[0], metadata=data() | {'case_id': 'BUG-000021'}) assert response.status_code == 422 assert not list(service.workspace.inbox.glob('*.zip')) raw = client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + TOKEN, 'Idempotency-Key': 'duplicate-json'}, data={'metadata': '{"kind":"new_case","kind":"case_return"}'}, files={'wowdiag': ('source', source_pair[0].read_bytes())}) assert raw.status_code == 422 first = post(client, source_pair[0]).json() stranger = Client('stranger', 'OTHER', ('other-project',)) with pytest.raises(APIError) as exc: service.status(first['submission_id'], stranger) assert exc.value.status == 404 def test_pair_and_attachments_are_retained(api, source_pair): client, service, _ = api response = client.post('/v1/submissions', headers={'Authorization': 'Bearer ' + TOKEN, 'Idempotency-Key': 'paired'}, data={'metadata': json.dumps(data())}, files=[ ('wowdiag', ('diag', source_pair[0].read_bytes())), ('wowsrc', ('src', source_pair[1].read_bytes())), ('attachments', ('a', b'opaque-human-notes'))]) assert response.status_code == 201, response.text with PackageReader(next(service.workspace.inbox.glob('*.zip'))) as reader: index = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_inputs'])) assert any(reader.read_bytes(ref) == b'opaque-human-notes' for ref in index['attachments'])