192 lines
11 KiB
Python
192 lines
11 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import ChangeRequestBuilder, OutputValidator, OutputError
|
|
from viano_bug_analysis.analysis.change_request_contracts import validate_request, make_document, render_human_request, ChangeRequestError
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.evidence.store import digest
|
|
from viano_bug_analysis.lifecycle import State, CheckpointStore, Workspace
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
from viano_bug_analysis.package import PackageReader, PackageWriter, PackageValidator
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_tools import tool_case
|
|
from test_change_request_builder import diagnosed, RequestAdapter, build
|
|
from test_investigator import CASE
|
|
|
|
|
|
def human():
|
|
return {'status':'assigned','title':'Collect approved placement data',
|
|
'reason':'The cause is established but remediation data is missing.',
|
|
'instructions':['Record approved positions and complete the template externally.'],
|
|
'expected_results':['Placement data and a completed remediation template for the next instance.']}
|
|
|
|
|
|
def template(payload, context):
|
|
payload.update(status='template',missing_information=['Approved placement coordinate.'],
|
|
placeholders=[{'token':'{{POSITION_X}}','description':'Approved horizontal coordinate.'}],
|
|
implementation_guidance=['Create a template using {{POSITION_X}}; do not guess the coordinate.'],
|
|
human_dev=human())
|
|
|
|
|
|
def no_coding(payload, context):
|
|
payload.update(status='none',affected_scope=[],validation_requirements=[],implementation_guidance=[],
|
|
required_change='Kein Codingauftrag vorhanden',implementation_intent='No coding assignment is needed.',
|
|
reason='Remediation is an external human task.',human_dev=human())
|
|
|
|
|
|
def test_template_and_human_data_task_are_completed_handoffs(diagnosed, tmp_path):
|
|
store, original=diagnosed
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
adapter=RequestAdapter(template);g=gateway(adapter)
|
|
result=build(ChangeRequestBuilder(store,g))
|
|
assert result.status=='COMPLETED' and result.failure_ref is None
|
|
with PackageReader(original) as old, PackageReader(result.package) as reader:
|
|
context=json.loads(reader.read_bytes(reader.manifest['artifacts']['change_request_context']))
|
|
assert context['parameter_grounding'] == 2
|
|
assert context['database_fact_bindings'] == [] # project-independent fixture
|
|
coding=json.loads(reader.read_bytes(reader.manifest['artifacts']['change_request']))
|
|
assert coding['assignment']=='assigned' and coding['request']['status']=='template'
|
|
assert coding['human_dev_assigned'] is True and 'human_dev' not in coding['request']
|
|
assert coding['request']['placeholders'][0]['token']=='{{POSITION_X}}'
|
|
text=reader.read_bytes(reader.manifest['artifacts']['human_dev_request']).decode()
|
|
assert 'Auftrag vorhanden' in text and 'complete the template externally' in text
|
|
assert not reader.manifest['artifacts']['failure_report']
|
|
for name in old.files:
|
|
if name!='manifest.json':assert reader.read_bytes(name)==old.read_bytes(name)
|
|
# No readiness/dependency or human-completion gate: the unresolved template
|
|
# and data task are exported immediately, with unchanged research budgets.
|
|
exported=asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
assert exported.status=='OUTPUT_READY' and exported.exported.parent==store.workspace.outbox
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
assert PackageValidator().validate(exported.exported).valid
|
|
portable=CheckpointStore(Workspace(tmp_path/'recipient'));portable.intake(exported.exported)
|
|
assert asyncio.run(OutputValidator(portable,g).publish(*CASE)).status=='OUTPUT_READY'
|
|
assert adapter.calls==1
|
|
store.advance(*CASE,State.HANDED_OFF,step_id='recipient-ack')
|
|
|
|
|
|
@pytest.mark.parametrize('human_assigned',[False,True])
|
|
def test_no_coding_assignment_still_exports_both_documents(diagnosed,human_assigned):
|
|
store,_=diagnosed
|
|
def change(p,c):
|
|
no_coding(p,c)
|
|
if not human_assigned:
|
|
p['human_dev']={'status':'none','title':'Kein Human-Dev-Auftrag vorhanden',
|
|
'reason':'No additional human work is needed.','instructions':[],'expected_results':[]}
|
|
adapter=RequestAdapter(change);g=gateway(adapter)
|
|
assert build(ChangeRequestBuilder(store,g)).status=='COMPLETED'
|
|
result=asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
assert result.status=='OUTPUT_READY'
|
|
with PackageReader(result.package) as reader:
|
|
doc=json.loads(reader.read_bytes(reader.manifest['artifacts']['change_request']))
|
|
assert doc['assignment']=='none' and doc['human_dev_assigned']==human_assigned
|
|
assert reader.read_bytes(reader.manifest['artifacts']['human_dev_request'])
|
|
assert reader.manifest['routing']['destination']=='outbox'
|
|
|
|
|
|
def test_missing_human_document_pointer_prevents_export(diagnosed,tmp_path):
|
|
store,_=diagnosed;g=gateway(RequestAdapter())
|
|
result=build(ChangeRequestBuilder(store,g));bad=tmp_path/'pointer.zip'
|
|
PackageWriter().write(result.package,bad,manifest_updates={'artifacts':{'human_dev_request':None}})
|
|
store.intake(bad)
|
|
with pytest.raises(OutputError):asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
assert not list(store.workspace.outbox.glob('*.zip'))
|
|
|
|
|
|
@pytest.mark.parametrize('legacy_insufficient',[False,True])
|
|
def test_legacy_pending_request_keeps_its_prompt_and_portable_replay(diagnosed,legacy_insufficient):
|
|
def legacy_change(p,c):
|
|
if legacy_insufficient:
|
|
p.update(status='insufficient_context',affected_scope=[],validation_requirements=[],
|
|
missing_information=['Approved placement data is missing.'])
|
|
store,_=diagnosed;adapter=RequestAdapter(legacy_change);g=gateway(adapter);builder=ChangeRequestBuilder(store,g)
|
|
with PackageReader(store.latest(*CASE)) as reader:context=builder.context(reader,3)
|
|
ref='analysis/change-requests/0001/context.json';raw=json_bytes(context)
|
|
store.advance(*CASE,State.BUILDING_DIAGNOSIS,step_id='change-request:v1:1:start',
|
|
details={'operation':'change-request-start','context_ref':ref,'context_sha256':digest(context)},
|
|
additions={ref:raw},artifacts={'change_request_context':ref})
|
|
result=build(builder)
|
|
assert result.status=='COMPLETED' and build(builder).reused and adapter.calls==1
|
|
with PackageReader(result.package) as reader:
|
|
doc=json.loads(reader.read_bytes(result.change_request_ref))
|
|
assert doc['format_version']==2 and reader.read_bytes(ref)==raw
|
|
assert reader.manifest['artifacts']['human_dev_request']
|
|
assert doc['assignment']==('none' if legacy_insufficient else 'assigned')
|
|
assert doc['human_dev_assigned']==legacy_insufficient
|
|
assert reader.manifest['state']==State.BUILDING_DIAGNOSIS
|
|
assert asyncio.run(OutputValidator(store,g).publish(*CASE)).status=='OUTPUT_READY'
|
|
assert adapter.calls==1
|
|
|
|
|
|
def minimal():
|
|
context={'handoff_protocol':2,'local_context':{'code.cpp':{'kind':'source'}},
|
|
'evidence':{'E000001':{'record':{'content_ref':'capture.json'},'content':{'items':[{'source_ref':'code.cpp'}]}}},
|
|
'diagnosis':{'diagnosis':{'root_cause':{'evidence_refs':['E000001']}}}}
|
|
payload={'status':'ready','problem_summary':'Observed defect','affected_scope':[{'kind':'source','source_ref':'code.cpp','evidence_refs':['E000001']}],
|
|
'required_change':'Correct the defect','implementation_intent':'Bounded correction',
|
|
'implementation_guidance':['Use the verified context'],'constraints':[],'preserve':[],'forbidden_changes':[],
|
|
'do_not_fix_by':[],'validation_requirements':[{'description':'Expected behavior restored','evidence_refs':['E000001']}],
|
|
'missing_information':[],'reason':'Sufficient coding context','placeholders':[],
|
|
'human_dev':{'status':'none','title':'No human work','reason':'No human task is needed','instructions':[],'expected_results':[]}}
|
|
return payload,context
|
|
|
|
|
|
@pytest.mark.parametrize('change,code',[
|
|
(lambda p: p.update(status='template',missing_information=['Coordinate']), 'PLACEHOLDERS'),
|
|
(lambda p: p.update(status='none'), 'NO_TASK_CONTRADICTORY'),
|
|
(lambda p: p['human_dev'].update(status='assigned'), 'HUMAN_DEV_INCOMPLETE'),
|
|
(lambda p: p['human_dev'].update(instructions=['Unexpected hidden work']), 'HUMAN_DEV_CONTRADICTORY'),
|
|
(lambda p: p.update(placeholders=[{'token':'{{X}}','description':'Value'}]), 'PLACEHOLDERS'),
|
|
])
|
|
def test_inconsistent_handoff_rejected(change,code):
|
|
payload,context=minimal();change(payload)
|
|
with pytest.raises(ChangeRequestError,match=code):validate_request(payload,context)
|
|
|
|
|
|
def test_coding_fields_do_not_contain_human_task_body():
|
|
payload,context=minimal();payload['human_dev']=human()
|
|
context.update(case_id='case',package_id='package',analysis_revision=1,diagnosis_ref='diagnosis.json',diagnosis_sha256='0'*64,
|
|
interpretation={'observed_behavior':[],'expected_behavior':[]})
|
|
context['local_context']['code.cpp'].update(source_ref='code.cpp',sha256='0'*64,size=1)
|
|
doc=make_document(context,payload,'context.json','call.json')
|
|
assert doc['human_dev_assigned']
|
|
assert 'Record approved positions' not in json.dumps(doc)
|
|
assert 'Record approved positions' in render_human_request(context,payload)
|
|
|
|
|
|
def test_verified_project_capture_can_ground_a_database_template():
|
|
payload,context=minimal()
|
|
template(payload,context)
|
|
payload['affected_scope']=[{'kind':'database','source_ref':'capture.json','evidence_refs':['E000001']}]
|
|
context['local_context']={'capture.json':{'kind':'database'}}
|
|
context['evidence']['E000001']['content']['items']=[{'query':{'kind':'database.rows'},'result':{'items':[{'row':{'id':1}}]}}]
|
|
validate_request(payload,context)
|
|
|
|
|
|
|
|
def test_independent_human_title_patch_survives_output_export_and_replay(diagnosed,tmp_path):
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.llm.answer_recovery import PATCH_PROMPT
|
|
store,_=diagnosed
|
|
def bad_title(payload,context):
|
|
template(payload,context);payload['human_dev']['title']=''
|
|
class Adapter(RequestAdapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
if request['prompt_version']==PATCH_PROMPT:
|
|
self.calls+=1
|
|
return ProviderResponse({'replacement':'Collect approved placement data'},'fixture',10,5)
|
|
return await super().generate(config,request,**kwargs)
|
|
adapter=Adapter(bad_title);g=gateway(adapter);builder=ChangeRequestBuilder(store,g)
|
|
result=build(builder)
|
|
assert result.status=='COMPLETED' and adapter.calls==2
|
|
assert build(builder).reused and adapter.calls==2
|
|
exported=asyncio.run(OutputValidator(store,g).publish(*CASE))
|
|
assert exported.status=='OUTPUT_READY'
|
|
portable=CheckpointStore(Workspace(tmp_path/'recipient'));portable.intake(exported.exported)
|
|
assert asyncio.run(OutputValidator(portable,g).publish(*CASE)).status=='OUTPUT_READY'
|
|
assert adapter.calls==2
|