87 lines
4.8 KiB
Python
87 lines
4.8 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from test_compact_investigation import context
|
|
from test_lean_investigation_decision import reply
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_project_research import source
|
|
from viano_bug_analysis.analysis.investigation_prompt import output_schema
|
|
from viano_bug_analysis.analysis.investigation_compact import validate_tool_shape
|
|
from viano_bug_analysis.analysis.investigation_config import InvestigationError
|
|
from viano_bug_analysis.llm.diagnostics import schema_feedback,sanitize
|
|
|
|
|
|
@pytest.mark.parametrize('kind',['source.search','source.symbol','source.registration','source.schema'])
|
|
def test_nested_enum_tool_feedback_names_missing_path_prefix(kind):
|
|
c=context();c['prompt_version']='investigation_project_v17';r=reply()
|
|
r['next']={'kind':'tool','operation':'project.query','arguments':{
|
|
'kind':kind,'text':'PRIVATE_QUERY','case_sensitive':False,'offset':0,'limit':20}}
|
|
with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r)
|
|
info=caught.value.feedback
|
|
assert info['validation_path']=='/next/arguments'
|
|
assert info['validation_fields']==['path_prefix']
|
|
assert 'path_prefix="" searches_repository_root' in info['requirement']
|
|
assert 'PRIVATE_QUERY' not in json.dumps(info)
|
|
|
|
|
|
def test_feedback_descends_through_nested_filters_and_root_definitions():
|
|
c=context();c['prompt_version']='investigation_project_v17';r=reply()
|
|
r['next']={'kind':'tool','operation':'project.query','arguments':{
|
|
'kind':'database.rows','table':'creature','columns':['guid'],
|
|
'filters':[{'column':'id','operator':'eq'}],'offset':0,'limit':20}}
|
|
with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r)
|
|
from viano_bug_analysis.analysis.tool_guidance import FILTER_REQUIREMENT
|
|
assert caught.value.feedback=={'validation_path':'/next/arguments/filters/0',
|
|
'validation_fields':['value'],'requirement':FILTER_REQUIREMENT}
|
|
|
|
|
|
def test_feedback_reports_type_error_without_recommending_missing_fields():
|
|
c=context();r=reply();r['next']={'kind':'tool','operation':'project.query','arguments':{
|
|
'kind':'source.search','text':'query','path_prefix':'','case_sensitive':False,'offset':0,'limit':'PRIVATE'}}
|
|
with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r)
|
|
assert caught.value.feedback['validation_path']=='/next/arguments/limit'
|
|
assert caught.value.feedback['requirement']=='match_expected_type_at_validation_path'
|
|
assert 'PRIVATE' not in json.dumps(caught.value.feedback)
|
|
|
|
|
|
def test_unknown_discriminator_does_not_choose_arbitrary_branch():
|
|
c=context();r=reply();r['next']={'kind':'tool','operation':'PRIVATE_OPERATION','arguments':{}}
|
|
error=next(Draft202012Validator(output_schema(c,strict=True)).iter_errors(r))
|
|
info=schema_feedback(error,include_keyword=True)
|
|
assert info['validation_path']=='/next' and info['validation_fields']==[]
|
|
assert sanitize(info)==info and 'PRIVATE' not in json.dumps(info)
|
|
|
|
|
|
def test_precise_feedback_reaches_checkpoint_then_one_corrected_source_query(tool_case,source):
|
|
from test_investigator import LoopAdapter,CASE,state
|
|
from test_case_interpreter import gateway
|
|
from viano_bug_analysis.analysis import Investigator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
store=tool_case[0]
|
|
with PackageReader(tool_case[1]) as reader:project=reader.manifest['project']
|
|
service=ProjectResearch(project,source=source);calls=[];original=source.query
|
|
def counted(args):calls.append(args);return original(args)
|
|
source.query=counted
|
|
def change(reply,c):
|
|
args={'kind':'source.search','text':'activate','case_sensitive':False,'offset':0,'limit':20}
|
|
if c.get('previous_rejection_details'):
|
|
assert c['previous_rejection_details']['validation_fields']==['path_prefix']
|
|
args['path_prefix']=''
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':args}
|
|
adapter=LoopAdapter(change);loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
first=asyncio.run(loop.step(*CASE));assert first.status=='REJECTED' and not calls
|
|
assert not BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['actions']
|
|
with PackageReader(first.package) as reader:
|
|
c=json.loads(reader.read_bytes(state(first)['active']['context_ref']))
|
|
assert c['previous_rejection_details']['validation_path']=='/next/arguments'
|
|
second=asyncio.run(loop.step(*CASE))
|
|
assert second.status=='PROGRESSED' and len(calls)==1
|
|
assert PackageValidator().validate(second.package).valid
|