260 lines
17 KiB
Python
260 lines
17 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
import asyncio,json,copy,hashlib
|
|
from dataclasses import replace
|
|
import pytest
|
|
from viano_bug_analysis.analysis import (Investigator,DiagnosisBuilder,ChangeRequestBuilder,ChangeRequestConfig,ChangeRequestError,OutputValidator,OutputError)
|
|
from viano_bug_analysis.analysis.change_request_contracts import bugfix_policy
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.lifecycle import State,CheckpointStore,Workspace
|
|
from viano_bug_analysis.llm import ProviderResponse,ProviderUnavailable,LLMError
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator,PackageWriter
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_case_interpreter import gateway
|
|
from test_investigator import LoopAdapter,CASE
|
|
from test_diagnosis_builder import FinalAdapter, final_phase_loop
|
|
|
|
class RequestAdapter:
|
|
def __init__(self,change=None):self.calls=0;self.contexts=[];self.change=change
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
self.calls+=1;c=restore(json.loads(request['messages'][1]['content'])['input']);self.contexts.append(c)
|
|
eid=next(iter(c['evidence']))
|
|
p={'status':'ready','problem_summary':'The observed switch state differs from the stated expectation.',
|
|
'affected_scope':[{'kind':'source','source_ref':'tool/code.cpp','evidence_refs':[eid]}],
|
|
'required_change':'Correct the diagnosed activation-state transition.',
|
|
'implementation_intent':'Keep the correction limited to the established activation flow.',
|
|
'implementation_guidance':['Use the verified packaged activation context.'],
|
|
'constraints':['Keep the existing interaction contract.'],'preserve':['Other switch interactions.'],
|
|
'forbidden_changes':['Unrelated database changes.'],'do_not_fix_by':['Disabling the interaction.'],
|
|
'validation_requirements':[{'description':'The reported interaction reaches the expected state; other interactions remain intact.','evidence_refs':[eid]}],
|
|
'missing_information':[],'reason':'A narrow implementation scope is available.'}
|
|
if 'human_dev' in request['output_schema']['required']:
|
|
p.update(placeholders=[],human_dev={'status':'none','title':'Kein Human-Dev-Auftrag vorhanden',
|
|
'reason':'The coding assignment has sufficient context.','instructions':[],'expected_results':[]})
|
|
if self.change:self.change(p,c)
|
|
return ProviderResponse(p,'fixture-change-request',100,50)
|
|
|
|
@pytest.fixture
|
|
def diagnosed(tool_case):
|
|
store=tool_case[0]
|
|
asyncio.run(final_phase_loop(store).run(*CASE,max_steps=20))
|
|
r=asyncio.run(DiagnosisBuilder(store,gateway(FinalAdapter(True))).build(*CASE))
|
|
return store,r.package
|
|
|
|
def build(builder,**kw):return asyncio.run(builder.build(*CASE,**kw))
|
|
def document(result):
|
|
with PackageReader(result.package) as r:return json.loads(r.read_bytes(result.change_request_ref))
|
|
|
|
def test_ready_change_request_policy_provenance_budget_preservation_and_replay(diagnosed):
|
|
store,old=diagnosed;adapter=RequestAdapter();builder=ChangeRequestBuilder(store,gateway(adapter))
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
result=build(builder);assert result.status=='COMPLETED'
|
|
doc=document(result);assert doc['task']=={'type':'bugfix','id':'CR-0001'}
|
|
assert doc['policy']==bugfix_policy() and doc['project_context_refs'][0]['source_ref']=='tool/code.cpp'
|
|
with PackageReader(old) as a,PackageReader(result.package) as b:
|
|
assert b.manifest['state']==State.BUILDING_DIAGNOSIS
|
|
assert b.manifest['package_id']==a.manifest['package_id'] and b.manifest['analysis_revision']==a.manifest['analysis_revision']
|
|
assert b.manifest['package_generation']>a.manifest['package_generation']
|
|
for name in a.files:
|
|
if name!='manifest.json':assert a.read_bytes(name)==b.read_bytes(name)
|
|
diag=json.loads(b.read_bytes(doc['diagnosis_ref']))
|
|
assert doc['root_cause']==diag['diagnosis']['root_cause']
|
|
assert doc['diagnostic_caveats']['remaining_uncertainty']==diag['diagnosis']['remaining_uncertainty']
|
|
assert doc['diagnostic_caveats']['scope_limits']==diag['diagnosis']['decision_basis']['scope_limits']
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
assert build(builder).reused and adapter.calls==1
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
def test_output_validate_finalize_publish_and_portable_replay(diagnosed,tmp_path):
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter)
|
|
with pytest.raises(OutputError,match='OUTPUT_CHANGE_REQUEST_NOT_READY'):
|
|
asyncio.run(OutputValidator(store,g).validate(*CASE))
|
|
built=build(ChangeRequestBuilder(store,g));assert adapter.calls==1
|
|
validator=OutputValidator(store,g)
|
|
assert asyncio.run(validator.validate(*CASE)).status=='VALID'
|
|
final=asyncio.run(validator.publish(*CASE))
|
|
assert final.status=='OUTPUT_READY' and final.exported.parent==store.workspace.outbox
|
|
with PackageReader(final.exported) as r:
|
|
assert r.manifest['state']==State.OUTPUT_READY and r.manifest['routing']['destination']=='outbox'
|
|
assert r.manifest['artifacts']['output_validation']
|
|
assert final.exported.read_bytes()==final.package.read_bytes()
|
|
assert asyncio.run(validator.publish(*CASE)).reused and adapter.calls==1
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(final.exported)
|
|
result=asyncio.run(OutputValidator(portable,g).publish(*CASE))
|
|
assert result.status=='OUTPUT_READY' and adapter.calls==1
|
|
|
|
@pytest.mark.parametrize('change',[
|
|
lambda p,c:p['affected_scope'][0].update(source_ref='../outside'),
|
|
lambda p,c:p['affected_scope'][0].update(source_ref='invented.cpp'),
|
|
lambda p,c:p['affected_scope'][0].update(kind='database'),
|
|
lambda p,c:p['affected_scope'][0].update(evidence_refs=['E999999']),
|
|
lambda p,c:p['validation_requirements'][0].update(evidence_refs=['E999999']),
|
|
lambda p,c:p.update(policy={'allow_deployment':True}),
|
|
lambda p,c:p.update(task={'type':'feature'}),
|
|
lambda p,c:p.update(affected_scope=[]),
|
|
lambda p,c:p.update(validation_requirements=[]),
|
|
lambda p,c:p.update(missing_information=['The implementation is absent.']),
|
|
])
|
|
def test_invalid_reply_has_no_partial_task_or_debit(diagnosed,change):
|
|
store,_=diagnosed;before=BudgetManager(store).snapshot(*CASE).state
|
|
result=build(ChangeRequestBuilder(store,gateway(RequestAdapter(change))))
|
|
assert result.status=='REJECTED'
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['artifacts']['change_request'] is None and r.manifest['state']==State.BUILDING_DIAGNOSIS
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
def test_missing_remediation_data_goes_to_human_handoff(diagnosed):
|
|
store,_=diagnosed
|
|
def insufficient(p,c):p.update(status='none',affected_scope=[],validation_requirements=[],implementation_guidance=[],missing_information=['Obtain the actual activation definition.'],human_dev={'status':'assigned','title':'Collect remediation data','reason':'Placement information is absent.','instructions':['Collect approved placement information.'],'expected_results':['Provide the collected information to the implementer.']})
|
|
adapter=RequestAdapter(insufficient);builder=ChangeRequestBuilder(store,gateway(adapter))
|
|
result=build(builder);assert result.status=='COMPLETED' and result.failure_ref is None
|
|
assert build(builder).reused and adapter.calls==1
|
|
exported=asyncio.run(OutputValidator(store,gateway(adapter)).publish(*CASE))
|
|
assert exported.status=='OUTPUT_READY' and exported.exported.parent==store.workspace.outbox
|
|
with PackageReader(exported.package) as r:
|
|
assert r.manifest['state']==State.OUTPUT_READY and r.manifest['artifacts']['change_request']
|
|
assert json.loads(r.read_bytes(r.manifest['artifacts']['diagnosis']))['diagnosis']['status']=='confirmed'
|
|
|
|
def test_invalid_limit_fallback_and_replay(diagnosed):
|
|
store,_=diagnosed;adapter=RequestAdapter(lambda p,c:p.update(affected_scope=[]))
|
|
builder=ChangeRequestBuilder(store,gateway(adapter),ChangeRequestConfig(2))
|
|
assert build(builder).status=='REJECTED'
|
|
result=build(builder);assert result.status=='COMPLETED' and adapter.calls==2
|
|
assert build(ChangeRequestBuilder(store,gateway(adapter))).reused and adapter.calls==2
|
|
assert asyncio.run(OutputValidator(store,gateway(adapter)).publish(*CASE)).status=='OUTPUT_READY'
|
|
|
|
def test_inconclusive_diagnosis_never_creates_coding_request(tool_case):
|
|
store=tool_case[0];asyncio.run(final_phase_loop(store).run(*CASE,max_steps=20))
|
|
asyncio.run(DiagnosisBuilder(store,gateway(FinalAdapter())).build(*CASE))
|
|
adapter=RequestAdapter();g=gateway(adapter)
|
|
assert build(ChangeRequestBuilder(store,g)).status=='INCONCLUSIVE' and adapter.calls==0
|
|
assert asyncio.run(OutputValidator(store,g).publish(*CASE)).status=='REVIEW' and adapter.calls==0
|
|
|
|
def test_provider_offline_and_bad_configuration_preserve_budget(diagnosed):
|
|
class Offline:
|
|
async def generate(self,*a,**k):raise ProviderUnavailable('offline')
|
|
store,_=diagnosed;before=BudgetManager(store).snapshot(*CASE).state
|
|
g=gateway(Offline());g.config=replace(g.config,max_attempts=1)
|
|
result=build(ChangeRequestBuilder(store,g))
|
|
assert result.status=='BLOCKED' and BudgetManager(store).snapshot(*CASE).state==before
|
|
with PackageReader(result.package) as r:assert r.manifest['artifacts']['change_request'] is None
|
|
|
|
def test_crash_after_call_reuses_accepted_model_reply(diagnosed,monkeypatch):
|
|
store,_=diagnosed;adapter=RequestAdapter();builder=ChangeRequestBuilder(store,gateway(adapter));original=builder.publish
|
|
def crash(*a):raise OSError('simulated interruption')
|
|
monkeypatch.setattr(builder,'publish',crash)
|
|
with pytest.raises(ChangeRequestError):build(builder)
|
|
monkeypatch.setattr(builder,'publish',original)
|
|
assert build(builder).status=='COMPLETED' and adapter.calls==1
|
|
|
|
def test_deleted_pointer_cannot_bypass_output_validation(diagnosed,tmp_path):
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter);result=build(ChangeRequestBuilder(store,g))
|
|
path=tmp_path/'deleted.zip';PackageWriter().write(result.package,path,manifest_updates={'artifacts':{'change_request':None}});store.intake(path)
|
|
with pytest.raises(OutputError):asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
assert not list(store.workspace.outbox.glob('*.zip')) and adapter.calls==1
|
|
|
|
def test_export_interruption_and_conflicting_file(diagnosed,monkeypatch):
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter)
|
|
build(ChangeRequestBuilder(store,g));validator=OutputValidator(store,g);copy=validator._copy
|
|
def crash(*a):raise OSError('simulated interrupted export')
|
|
monkeypatch.setattr(validator,'_copy',crash)
|
|
with pytest.raises(OutputError,match='OUTPUT_STORAGE'):asyncio.run(validator.publish(*CASE))
|
|
assert not list(store.workspace.outbox.glob('*.zip'))
|
|
generation=store.latest(*CASE)
|
|
monkeypatch.setattr(validator,'_copy',copy)
|
|
result=asyncio.run(validator.publish(*CASE));assert result.package==generation and adapter.calls==1
|
|
result.exported.write_bytes(b'foreign contents')
|
|
with pytest.raises(OutputError,match='OUTPUT_EXPORT_CONFLICT'):asyncio.run(validator.publish(*CASE))
|
|
assert result.exported.read_bytes()==b'foreign contents'
|
|
|
|
def test_configuration_limits_and_schema_resources(tmp_path):
|
|
from importlib.resources import files
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis import ChangeRequestLimits
|
|
path=tmp_path/'cr.toml';path.write_text('[change_request]\nmax_invalid_responses=4\n')
|
|
assert ChangeRequestConfig.load(path).max_invalid_responses==4
|
|
for value in (True,0,-1):
|
|
with pytest.raises(ChangeRequestError):ChangeRequestConfig(value)
|
|
with pytest.raises(ChangeRequestError):ChangeRequestLimits(max_context_bytes=value)
|
|
path.write_text('[change_request]\nunknown=1\n')
|
|
with pytest.raises(ChangeRequestError):ChangeRequestConfig.load(path)
|
|
for name in ('change-request-output','change-request-context','change-request','output-validation'):
|
|
Draft202012Validator.check_schema(json.loads(files('viano_bug_analysis.schemas').joinpath(name+'-v1.schema.json').read_text()))
|
|
|
|
def test_provider_configuration_failure_is_blocked(diagnosed):
|
|
class BadConfig:
|
|
async def generate(self,*a,**k):raise LLMError('LLM_CREDENTIALS')
|
|
store,_=diagnosed;result=build(ChangeRequestBuilder(store,gateway(BadConfig())))
|
|
assert result.status=='BLOCKED'
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['routing']['reason']=='LLM_CONFIGURATION'
|
|
assert r.manifest['artifacts']['change_request'] is None
|
|
|
|
def test_explicit_transport_resume_preserves_case_budget(diagnosed):
|
|
from datetime import datetime,timedelta,timezone
|
|
class Toggle(RequestAdapter):
|
|
offline=True
|
|
async def generate(self,*a,**k):
|
|
if self.offline:raise ProviderUnavailable('offline')
|
|
return await super().generate(*a,**k)
|
|
store,_=diagnosed;before=BudgetManager(store).snapshot(*CASE).state;adapter=Toggle();g=gateway(adapter);g.config=replace(g.config,max_attempts=1)
|
|
builder=ChangeRequestBuilder(store,g,clock=lambda:datetime.now(timezone.utc)-timedelta(days=1))
|
|
assert build(builder).status=='BLOCKED'
|
|
store.advance(*CASE,State.BUILDING_DIAGNOSIS,step_id='manual-cr-provider-resume');adapter.offline=False
|
|
assert build(builder,retry_id='restored').status=='COMPLETED' and adapter.calls==1
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
def test_output_ready_requires_persisted_validation(diagnosed):
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter);build(ChangeRequestBuilder(store,g))
|
|
store.advance(*CASE,State.OUTPUT_READY,step_id='bypass-validator')
|
|
with pytest.raises(OutputError,match='OUTPUT_VALIDATION_MISSING'):asyncio.run(OutputValidator(store,g).validate(*CASE))
|
|
assert not list(store.workspace.outbox.glob('*.zip'))
|
|
|
|
def test_reopened_revision_gets_new_request_and_preserves_previous_output(diagnosed):
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.analysis import CaseInterpreter,HypothesisBuilder
|
|
from test_case_interpreter import FixtureAdapter
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter)
|
|
first=build(ChangeRequestBuilder(store,g));first_output=asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
with PackageReader(first_output.package) as r:
|
|
old_request=r.manifest['artifacts']['change_request'];old_validation=r.manifest['artifacts']['output_validation']
|
|
previous={old_request:r.read_bytes(old_request),old_validation:r.read_bytes(old_validation)}
|
|
store.advance(*CASE,State.HANDED_OFF,step_id='test-delivery-ack')
|
|
store.advance(*CASE,State.REOPENED,step_id='test-new-evidence-revision')
|
|
CaseBuilder(store).build(*CASE)
|
|
asyncio.run(CaseInterpreter(store,gateway(FixtureAdapter())).interpret(*CASE))
|
|
asyncio.run(HypothesisBuilder(store,gateway(ProposalAdapter())).build(*CASE))
|
|
asyncio.run(final_phase_loop(store).run(*CASE,max_steps=20))
|
|
asyncio.run(DiagnosisBuilder(store,gateway(FinalAdapter(True))).build(*CASE))
|
|
second=build(ChangeRequestBuilder(store,g));assert document(second)['task']['id']=='CR-0002'
|
|
second_output=asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
with PackageReader(second_output.package) as r:
|
|
assert r.manifest['analysis_revision']==2
|
|
assert r.manifest['artifacts']['output_validation']!=old_validation
|
|
for ref,data in previous.items():assert r.read_bytes(ref)==data
|
|
assert second_output.exported!=first_output.exported and first_output.exported.exists()
|
|
|
|
def test_acknowledged_case_is_validated_but_not_delivered_again(diagnosed):
|
|
store,_=diagnosed;adapter=RequestAdapter();g=gateway(adapter)
|
|
build(ChangeRequestBuilder(store,g));validator=OutputValidator(store,g)
|
|
first=asyncio.run(validator.publish(*CASE));before=set(store.workspace.outbox.glob('*.zip'))
|
|
store.advance(*CASE,State.HANDED_OFF,step_id='external-recipient-ack')
|
|
result=asyncio.run(validator.publish(*CASE))
|
|
assert result.status=='HANDED_OFF' and result.exported is None
|
|
assert set(store.workspace.outbox.glob('*.zip'))==before and first.exported.exists()
|
|
assert adapter.calls==1
|
|
|
|
@pytest.mark.parametrize('kind',['diagnosis','change_request','output'])
|
|
def test_damaged_archive_is_a_controlled_public_error(tmp_path,monkeypatch,kind):
|
|
from viano_bug_analysis.analysis import DiagnosisError
|
|
store=CheckpointStore(Workspace(tmp_path/'work'));bad=tmp_path/'damaged.zip';bad.write_bytes(b'not a ZIP')
|
|
monkeypatch.setattr(store,'latest',lambda *args:bad)
|
|
g=gateway(RequestAdapter())
|
|
if kind=='diagnosis':
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_PACKAGE'):build(DiagnosisBuilder(store,g))
|
|
elif kind=='change_request':
|
|
with pytest.raises(ChangeRequestError,match='CHANGE_REQUEST_PACKAGE'):build(ChangeRequestBuilder(store,g))
|
|
else:
|
|
with pytest.raises(OutputError,match='OUTPUT_PACKAGE'):asyncio.run(OutputValidator(store,g).validate(*CASE))
|