98 lines
4.7 KiB
Python
98 lines
4.7 KiB
Python
import copy
|
|
import json
|
|
import pytest
|
|
|
|
from test_investigation_lean import lean_context
|
|
from test_lean_investigation_decision import reply
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from viano_bug_analysis.analysis import InvestigationError
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from viano_bug_analysis.analysis.investigation_compact import validate_tool_shape
|
|
from viano_bug_analysis.analysis.tool_guidance import HEADER,FILTER_REQUIREMENT,filter_feedback
|
|
|
|
|
|
def context(lean=False):
|
|
c=lean_context()
|
|
if lean:c['lean_protocol']=2
|
|
else:c.pop('lean_session')
|
|
return c
|
|
|
|
|
|
@pytest.mark.parametrize('lean',[False,True])
|
|
@pytest.mark.parametrize('filters',[[],['column','id','operator','eq','value',23487]])
|
|
def test_recorded_filter_failures_get_exact_value_free_repair_and_valid_shape_passes(lean,filters):
|
|
c=context(lean)
|
|
action={'kind':'tool','operation':'project.query','arguments':{
|
|
'kind':'database.rows','table':'objects','columns':['id'],
|
|
'filters':filters,'offset':0,'limit':5}}
|
|
if lean:r={'assessment':None,'observation':'Scoped observation.','next':action}
|
|
else:
|
|
r=reply();r['next']=action;r['update']['next_step']=None
|
|
r['update']['working_brief']['findings']=[]
|
|
before=copy.deepcopy(r)
|
|
with pytest.raises(InvestigationError) as error:validate_tool_shape(c,r)
|
|
assert r==before
|
|
feedback=error.value.feedback
|
|
assert feedback['requirement']==FILTER_REQUIREMENT and len(FILTER_REQUIREMENT)<=160
|
|
assert '23487' not in json.dumps(feedback)
|
|
fixed=copy.deepcopy(r);fixed['next']['arguments']['filters']=[{'column':'id','operator':'eq','value':23487}]
|
|
validate_tool_shape(c,fixed)
|
|
|
|
|
|
@pytest.mark.parametrize('lean',[False,True])
|
|
def test_explicit_argument_shapes_once_in_full_and_slim_requests(lean):
|
|
c=context(lean);c['tool_guidance_version']=1
|
|
b=c['tool_schema']['x-project-bindings'];b['relation_profiles']={'inspect':{}};b['source_analysis']={'version':1}
|
|
request=build_prompt(c);system=request['messages'][0]['content']
|
|
assert system.count(HEADER)==1
|
|
assert '"kind":"database.relations"' in system and '"profile"' in system
|
|
assert '"kind":"source.context"' in system and '"path_prefix":""' in system
|
|
assert 'nonempty array of objects' in system and 'No table or SQL argument' in system
|
|
# Already stored contexts without the new marker keep their old messages.
|
|
c.pop('tool_guidance_version')
|
|
assert HEADER not in build_prompt(c)['messages'][0]['content']
|
|
|
|
|
|
def test_no_new_research_instructions_during_consolidation():
|
|
c=context(True);c['tool_guidance_version']=1;c['lean_session']['checkpoint']=True
|
|
c['tool_schema']['x-project-bindings']['relation_profiles']={'inspect':{}}
|
|
assert HEADER not in build_prompt(c)['messages'][0]['content']
|
|
|
|
|
|
def test_saved_generic_feedback_is_not_retroactively_rewritten():
|
|
from viano_bug_analysis.analysis.investigation_lean import repair_feedback
|
|
old={'validation_path':'/next/arguments/filters/0','validation_fields':[],
|
|
'requirement':'match_expected_type_at_validation_path'}
|
|
assert repair_feedback(old)==old
|
|
assert filter_feedback(old)['requirement']==FILTER_REQUIREMENT
|
|
nested={'validation_path':'/next/arguments/filters/0/operator','requirement':'use_allowed_value_at_validation_path'}
|
|
assert filter_feedback(nested)==nested
|
|
|
|
|
|
def test_feedback_persists_then_fixture_model_corrects_without_extra_tool_actions(tool_case):
|
|
from test_working_brief import service_for
|
|
from test_investigator import LoopAdapter,run,state
|
|
from test_case_interpreter import gateway
|
|
from test_project_research import rows
|
|
from viano_bug_analysis.analysis import Investigator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageReader
|
|
attempts=[]
|
|
def change(r,c):
|
|
if c.get('previous_rejection_details'):
|
|
assert c['previous_rejection_details']['requirement']==FILTER_REQUIREMENT
|
|
q=rows()
|
|
if not attempts:q['filters']=[]
|
|
attempts.append(q)
|
|
r['next']={'kind':'tool','operation':'project.query','arguments':q}
|
|
adapter=LoopAdapter(change)
|
|
loop=Investigator(tool_case[0],gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
first=run(loop,steps=1)
|
|
assert state(first)['hypotheses']['H1']['invalid_responses']==1
|
|
second=run(loop,steps=1)
|
|
assert len(attempts)==2 and state(second)['active'] is None
|
|
assert state(second)['hypotheses']['H1']['invalid_responses']==0
|
|
with PackageReader(second.package) as reader:
|
|
assert len([p for p in reader.files if p.startswith('analysis/evidence/captures/')])==1
|