174 lines
10 KiB
Python
174 lines
10 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from viano_bug_analysis.tools.repository import GitSourceProvider
|
|
from viano_bug_analysis.tools.source_relevance import case_seeds
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
root=tmp_path/'repo'
|
|
for d in ['src','sql/base','sql/old']: (root/d).mkdir(parents=True,exist_ok=True)
|
|
(root/'src/a_noise.cpp').write_text('\n'.join('SPELL_NOISE_'+str(i)+' = 77;' for i in range(20))+'\nvoid AddSC_noise() {}\n',encoding='utf-8')
|
|
(root/'src/target.cpp').write_text('enum { NPC_TARGET = 77 };\nvoid npc_target() { summon(NPC_TARGET); }\nvoid AddSC_target() { npc_target(); }\n',encoding='utf-8')
|
|
(root/'src/loader.cpp').write_text('void AddSC_target();\nvoid load() { AddSC_target(); }\nvoid AddSC_unrelated();\n',encoding='utf-8')
|
|
(root/'src/coordinates.cpp').write_text('float x = 77.123;\nint larger = 177;\nint random = 88;\n',encoding='utf-8')
|
|
(root/'src/other_area.cpp').write_text('void npc_cross_zone() {}\n',encoding='utf-8')
|
|
(root/'src/zone.cpp').write_text('void npc_local() {}\n',encoding='utf-8')
|
|
(root/'sql/base/world.sql').write_text('CREATE TABLE target (id int);\n',encoding='utf-8')
|
|
(root/'sql/old/noise.sql').write_text('INSERT INTO spell VALUES (77);\n',encoding='utf-8')
|
|
def git(*args):return subprocess.run(['git','-C',str(root),*args],check=True,capture_output=True).stdout.decode().strip()
|
|
git('init');git('add','.');git('-c','user.name=Test','-c','user.email=test@example.invalid','commit','-m','fixture')
|
|
return GitSourceProvider(root,git('rev-parse','HEAD'),'https://example.invalid/repo')
|
|
|
|
|
|
def service(source,**settings):
|
|
return ProjectResearch({},source=source,source_analysis={'protocol':2,'max_searches':8,
|
|
'code_prefixes':['src/'],'schema_prefixes':['sql/base/'],'historical_prefixes':['sql/old/'],
|
|
'registration_prefixes':['AddSC_'],'type_markers':{'actor':['NPC'],'effect':['SPELL']},**settings})
|
|
|
|
|
|
def query(text='77',**extra):return {'kind':'source.context','text':text,'path_prefix':'',**extra}
|
|
|
|
|
|
def seeds(*anchors):return {'anchors':list(anchors),'omitted_anchors':0}
|
|
|
|
|
|
def anchor(term,kind='id',group='N0001:id',**extra):
|
|
return {'term':term,'kind':kind,'group':group,'source_ref':'case/case.json',
|
|
'json_pointer':'/entities/0','entity_type':'actor',**extra}
|
|
|
|
|
|
def test_typed_collision_not_frequency_selects_implementation(source):
|
|
r=json.loads(service(source).collect(query(),source_context=seeds(anchor('77'))))['result']
|
|
items={i['path']:i for i in r['items']}
|
|
assert items['src/target.cpp']['tier']=='primary_candidate'
|
|
assert items['src/a_noise.cpp']['tier']=='weak_numeric' and items['src/a_noise.cpp']['blocks']==[]
|
|
assert 'src/coordinates.cpp' not in items
|
|
assert all('different_or_ambiguous_identifier_namespace' in [s['reason'] for s in h['signals']] for h in items['src/a_noise.cpp']['hits'])
|
|
assert 'AddSC_noise' not in [s['query']['text'] for s in r['searches']]
|
|
assert {'NPC_TARGET','AddSC_target'} <= {s['query']['text'] for s in r['searches']}
|
|
|
|
|
|
def test_untyped_numbers_do_not_promote_or_expand_even_when_only_hits(source):
|
|
r=json.loads(service(source).collect(query()))['result']
|
|
assert r['selection']['primary_files']==[]
|
|
assert all(i['tier']=='weak_numeric' and not i['blocks'] for i in r['items'])
|
|
assert len(r['searches'])==1
|
|
assert len([h for i in r['items'] for h in i['hits']])<=8
|
|
|
|
|
|
def test_names_variants_are_not_independent_corroboration(source):
|
|
r=json.loads(service(source).collect(query('77'),source_context=seeds(
|
|
anchor('npc_target','name','same_name'),anchor('NPC_TARGET','name','same_name'))))['result']
|
|
item=next(i for i in r['items'] if i['path']=='src/target.cpp')
|
|
assert item['tier']=='supporting_context' and item['ranking_score']==25
|
|
r=json.loads(service(source).collect(query('npc_target'),source_context=seeds(
|
|
anchor('npc_target','name','same_name'))))['result']
|
|
assert r['selection']['primary_files']==[]
|
|
|
|
|
|
def test_known_other_area_remains_usable_context(source):
|
|
r=json.loads(service(source,location_paths={'Home':['src/zone.cpp'],'Elsewhere':['src/other_area.cpp']}).collect(
|
|
query('npc_local'),source_context=seeds(anchor('npc_cross_zone','script','script'),
|
|
anchor('Home','context','place'))))['result']
|
|
items={i['path']:i for i in r['items']}
|
|
assert items['src/other_area.cpp']['tier']=='supporting_other_location'
|
|
assert items['src/other_area.cpp']['blocks']
|
|
assert 'src/other_area.cpp' in r['selection']['supporting_files']
|
|
|
|
|
|
def test_scopes_do_not_silently_mix_code_schema_and_history(source):
|
|
s=service(source)
|
|
assert not any(i['path'].endswith('.sql') for i in json.loads(s.collect(query()))['result']['items'])
|
|
r=json.loads(s.collect(query('CREATE TABLE',scope='schema')))['result']
|
|
assert [i['path'] for i in r['items']]==['sql/base/world.sql']
|
|
r=json.loads(s.collect(query(scope='historical_sql')))['result']
|
|
assert [i['path'] for i in r['items']]==['sql/old/noise.sql']
|
|
with pytest.raises(ToolError,match='SCOPE_CONFLICT'):s.collect(query(path_prefix='sql/old/'))
|
|
with pytest.raises(ToolError):s.collect(query(scope='all'))
|
|
|
|
|
|
def test_new_args_require_a_case_and_do_not_accept_model_fabricated_anchors(source):
|
|
s=service(source)
|
|
with pytest.raises(ToolError,match='CONTEXT_REQUIRED'):s.collect(query(entity_refs=['N0001']))
|
|
with pytest.raises(ToolError):s.collect({**query(),'anchors':[anchor('npc_fake','script')]})
|
|
assert ProjectResearch({},source=source,source_analysis={}).source_context.catalog()['version']==1
|
|
with pytest.raises(ToolError):ProjectResearch({},source=source,source_analysis={'code_prefixes':['src/']})
|
|
|
|
|
|
def test_selection_limits_are_explicit_and_source_worktree_is_ignored(source):
|
|
(source.path/'src/target.cpp').write_text('wrong mutable worktree\n',encoding='utf-8')
|
|
r=json.loads(service(source,max_total_lines=1).collect(query(),source_context=seeds(anchor('77'))))['result']
|
|
assert r['truncated'] and 'line_limit' in r['limits_reached']
|
|
assert any(i.get('omitted_context_lines',0)>0 for i in r['items'])
|
|
assert 'wrong mutable' not in json.dumps(r)
|
|
|
|
|
|
@pytest.mark.parametrize('settings',[{'type_markers':{'actor':['.*']}},{'protocol':True},
|
|
{'code_prefixes':['../escape']},{'max_seed_terms':0},{'location_paths':{'Home':['C:/bad']}}])
|
|
def test_invalid_adapter_rules_fail_controlled(source,settings):
|
|
with pytest.raises(ToolError):service(source,**settings)
|
|
|
|
|
|
def test_case_collector_names_roles_scripts_and_provenance(source,monkeypatch):
|
|
import viano_bug_analysis.case.contracts as contracts
|
|
sources={'N0001':{'kind':'entity','data':{'type':'actor','id':'77','name':'Target actor','roles':['affected_subject']}},
|
|
'N0002':{'kind':'entity','data':{'type':'effect','id':'88','name':'Related quest','roles':['selected_focus']}}}
|
|
captured={'query':{'kind':'database.rows','table':'actors'},'result':{'items':[{'row':{'entry':77,'ScriptName':'npc_target'}}]}}
|
|
fact={'id':'F0001','name':'project_query','value':{'capture_ref':'research/capture.json',**captured}}
|
|
case={'runtime_facts':{'facts':[fact,{'id':'F0002','name':'location_name','value':'Home'}]}}
|
|
ctx=SimpleNamespace(case=case,case_ref='case/case.json',reader=SimpleNamespace(
|
|
manifest={'artifacts':{'interpretation_context':'case/context.json'}},files=['research/capture.json']))
|
|
monkeypatch.setattr(contracts,'read_json',lambda r,ref,*args: {'sources':sources} if ref=='case/context.json' else captured)
|
|
s=service(source,focus_roles=['selected_focus'],context_fact_names=['location_name'],entity_tables=[
|
|
{'table':'actors','entity_type':'actor','id_column':'entry','script_columns':['ScriptName']}])
|
|
c=case_seeds(ctx,query(),s.source_context.config)
|
|
assert c['entity_refs']==['N0001','N0002']
|
|
assert any(a['term']=='npc_target' and a['source_ref']=='research/capture.json' for a in c['anchors'])
|
|
assert any(a['term']=='Home' for a in c['anchors'])
|
|
assert all(a['source_ref'] and a['json_pointer'] for a in c['anchors'])
|
|
with pytest.raises(ToolError,match='ENTITY_REF'):case_seeds(ctx,query(entity_refs=['N9999']),s.source_context.config)
|
|
|
|
|
|
def test_ambiguous_id_does_not_inherit_a_namespace(source,monkeypatch):
|
|
import viano_bug_analysis.case.contracts as contracts
|
|
sources={sid:{'kind':'entity','data':{'type':typ,'id':'77','roles':[]}}
|
|
for sid,typ in [('N0001','actor'),('N0002','effect')]}
|
|
monkeypatch.setattr(contracts,'read_json',lambda *args:{'sources':sources})
|
|
ctx=SimpleNamespace(case={},case_ref='case/case.json',reader=SimpleNamespace(
|
|
manifest={'artifacts':{'interpretation_context':'case/context.json'}}))
|
|
s=service(source);c=case_seeds(ctx,query(),s.source_context.config)
|
|
assert c['ambiguous_identity'] and all(a.get('entity_type') is None for a in c['anchors'])
|
|
r=json.loads(s.collect(query(),source_context=c))['result']
|
|
assert r['selection']['primary_files']==[]
|
|
|
|
|
|
def test_v2_fixture_loop_captures_case_seeds_and_resumes(source,tool_case,tmp_path):
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_hybrid import HybridAdapter
|
|
from test_investigator import run
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
s=service(source)
|
|
with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project']
|
|
adapter=HybridAdapter();adapter.bootstrap.change=lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':query('npc_target')})
|
|
router=InvestigationToolRouter(project_research=s)
|
|
first=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),router=router),steps=3)
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
final=run(Investigator(portable,gateway(adapter),router=router))
|
|
assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
|
|
from viano_bug_analysis.llm.cache_measurement import measure_package
|
|
assert measure_package(final.package)['all_exactly_reconstructed']
|