import json import pytest from starlette.testclient import TestClient from test_case_builder import case_setup from test_case_interpreter import built_case from test_entity_resolution import Model,service_for,run from viano_bug_analysis.api import APIConfig,Project,Client from viano_bug_analysis.api.app import create_app from viano_bug_analysis.api.cases import case_key from viano_bug_analysis.lifecycle import CheckpointStore from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain from viano_bug_analysis.package import PackageReader from viano_bug_analysis.worker import AnalysisWorker,WorkerConfig,WorkerError from viano_bug_analysis.worker.repair import approve_entity_repair @pytest.fixture(autouse=True) def project_manifest(manifest): manifest['project']={'id':'p','type':'wow','profile':'vcore'} def exhaust(store): model=Model();model.invalid_plan=3;service=service_for(store) for _ in range(3):assert run(store,model,service).status=='REJECTED' assert run(store,model,service).status=='BLOCKED' return model,service def test_approval_keeps_revision_artifacts_and_only_allows_one_attempt(built_case,tmp_path): store=built_case[0];model,service=exhaust(store) with PackageReader(store.latest('case-A','stable-package')) as reader: manifest=reader.manifest;before={n:reader.read_bytes(n) for n in reader.files if n!='manifest.json'} path=approve_entity_repair(store,'case-A','stable-package',actor='alice',reason='Updated plan validation guidance',expected_generation=manifest['package_generation']) with PackageReader(path) as reader: assert reader.manifest['analysis_revision']==manifest['analysis_revision'] assert {k:v for k,v in reader.manifest['artifacts'].items() if k!='lifecycle_checkpoint'}=={k:v for k,v in manifest['artifacts'].items() if k!='lifecycle_checkpoint'} assert reader.manifest['state']=='INTERPRETING_CASE' for name,raw in before.items():assert reader.read_bytes(name)==raw details=checkpoint_chain(reader)[0][1]['details'] assert details['repair']['attempt']==4 and details['actor']=='alice' with pytest.raises(WorkerError): approve_entity_repair(store,'case-A','stable-package',actor='bob',reason='duplicate',expected_generation=manifest['package_generation']) # Another invalid response exhausts this approval; no new attempt is automatic. model.invalid_plan=1 assert run(store,model,service).status=='REJECTED' assert run(store,model,service).status=='BLOCKED' assert run(store,model,service).status=='BLOCKED' assert len(model.requests)==4 with PackageReader(store.latest('case-A','stable-package')) as reader:generation=reader.manifest['package_generation'] approve_entity_repair(store,'case-A','stable-package',actor='alice',reason='Second explicit correction',expected_generation=generation) result=run(store,model,service) assert result.status=='COMPLETED' recipient=CheckpointStore(type(store.workspace)(tmp_path/'portable')) recipient.intake(result.package) calls=len(model.requests) assert run(recipient,model).reused and len(model.requests)==calls def test_web_approval_is_scoped_csrf_protected_and_audited(built_case,monkeypatch): store=built_case[0];model,service=exhaust(store) monkeypatch.setenv('MACHINE_TOKEN','m_'+'x'*40) config=APIConfig(store.workspace.root,(Project('p','wow','vcore'),), (Client('capture','MACHINE_TOKEN',('p',)),)) app=create_app(config);web=TestClient(app) app.state.sessions.users.manage('create_user','alice','example-password') login=web.post('/v1/auth/login',json={'username':'alice','password':'example-password'}) headers={'X-CSRF-Token':login.json()['csrf']} with PackageReader(store.latest('case-A','stable-package')) as reader:generation=reader.manifest['package_generation'] request={'action':'repair','key':case_key('case-A','stable-package'),'package_generation':generation,'reason':'Installed technical correction'} assert web.post('/v1/operations',json=request).status_code==403 machine=TestClient(app) assert machine.post('/v1/operations',headers={'Authorization':'Bearer m_'+'x'*40},json=request).status_code==403 assert web.post('/v1/operations',headers=headers,json={**request,'reason':' '}).status_code==422 assert web.post('/v1/operations',headers=headers,json={**request,'package_generation':generation-1}).status_code==409 reply=web.post('/v1/operations',headers=headers,json=request) assert reply.status_code==200 and reply.json()['status']=='QUEUED' assert web.post('/v1/operations',headers=headers,json=request).json()['command_id']==reply.json()['command_id'] worker=AnalysisWorker(WorkerConfig(config,min_free_bytes=1)) worker._commands() assert not worker.control.read()['commands'] assert worker.control.read()['events'][-1]['action']=='repair' assert worker.control.read()['events'][-1]['status']=='APPLIED' assert run(store,model,service).status=='COMPLETED' def test_assessment_approval_keeps_plan_and_database_captures(built_case): store=built_case[0];model=Model();model.invalid_assessment=3;service=service_for(store) for _ in range(3):assert run(store,model,service).status=='REJECTED' assert run(store,model,service).status=='BLOCKED' with PackageReader(store.latest('case-A','stable-package')) as reader: generation=reader.manifest['package_generation'] plan=reader.read_bytes('analysis/interpretations/0001/entity-resolution/plan.json') search=reader.read_bytes('analysis/interpretations/0001/entity-resolution/search.json') approve_entity_repair(store,'case-A','stable-package',actor='alice',reason='Precise assessment repair installed',expected_generation=generation) result=run(store,model,service) assert result.status=='COMPLETED' and len(service.database.calls)==1 with PackageReader(result.package) as reader: assert reader.read_bytes('analysis/interpretations/0001/entity-resolution/plan.json')==plan assert reader.read_bytes('analysis/interpretations/0001/entity-resolution/search.json')==search @pytest.mark.parametrize('reason',['AWAITING_TEST_FEEDBACK','RESEARCH_UNAVAILABLE','INVALID_LLM_RESPONSES']) def test_other_blocks_cannot_receive_an_entity_allowance(built_case,reason): store=built_case[0] # The protected human-feedback state is rejected before its resume semantics matter. store.advance('case-A','stable-package','BLOCKED',step_id='other-block',reason=reason, retry_not_before='2099-01-01T00:00:00+00:00') with PackageReader(store.latest('case-A','stable-package')) as reader:generation=reader.manifest['package_generation'] with pytest.raises(WorkerError): approve_entity_repair(store,'case-A','stable-package',actor='alice',reason='Do not bypass',expected_generation=generation)