205 lines
11 KiB
Python
205 lines
11 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch, ProjectLimits
|
|
from viano_bug_analysis.tools.repository import GitSourceProvider
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
root=tmp_path/'source';(root/'src').mkdir(parents=True)
|
|
(root/'src/events.cpp').write_text('enum Ids {\nNPC_WILD = 23487,\n};\nvoid spawn() {\nsummon(NPC_WILD);\n}\nvoid AddSC_events() {\nregister_handler("npc_wild",spawn);\n}\n',encoding='utf-8')
|
|
(root/'src/loader.cpp').write_text('void AddSC_events();\nvoid load() { AddSC_events(); }\nvoid AddSC_unrelated();\n',encoding='utf-8')
|
|
(root/'src/unrelated.cpp').write_text('int NPC_WILD_LONGER;\n',encoding='utf-8')
|
|
(root/'src/module.py').write_text('def activate():\n return "python context"\n\nactivate()\n',encoding='utf-8')
|
|
(root/'src/long.cpp').write_text('LONG_TARGET '+('x'*3000)+'\n',encoding='utf-8')
|
|
(root/'src/gaps.cpp').write_text('GAPPED\n'+('\n'*25)+'GAPPED\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')
|
|
commit=git('rev-parse','HEAD')
|
|
(root/'src/events.cpp').write_text('mutable worktree content must be ignored\n',encoding='utf-8')
|
|
return GitSourceProvider(root,commit,'https://example.invalid/repo')
|
|
|
|
|
|
def service(source,**settings):
|
|
return ProjectResearch({},source=source,source_analysis={'registration_prefixes':['AddSC_'],**settings})
|
|
|
|
|
|
def query(text='23487',prefix='src/'):
|
|
return {'kind':'source.context','text':text,'path_prefix':prefix}
|
|
|
|
|
|
def test_chain_alias_registration_context_is_compact_and_pinned(source):
|
|
s=service(source);capture=json.loads(s.collect(query()));result=capture['result']
|
|
assert [x['query']['text'] for x in result['searches']]==['23487','NPC_WILD','AddSC_events']
|
|
assert not result['truncated']
|
|
assert {i['path'] for i in result['items']}=={'src/events.cpp','src/loader.cpp'}
|
|
assert all(len(i['blocks'])==1 for i in result['items'])
|
|
events=result['items'][0]
|
|
assert events['blocks'][0]['start_line']==1
|
|
assert 'NPC_WILD = 23487,' in events['blocks'][0]['lines']
|
|
assert 'mutable worktree' not in json.dumps(result)
|
|
assert capture['origin']['commit']==source.commit
|
|
assert 'not verified semantic' in result['interpretation']
|
|
assert 'AddSC_unrelated' not in [s['query']['text'] for s in result['searches']]
|
|
assert s.collect(query())==s.collect(query())
|
|
|
|
|
|
def test_python_and_empty_scoped_search(source):
|
|
s=service(source)
|
|
r=json.loads(s.collect(query('activate','')))['result']
|
|
assert [i['path'] for i in r['items']]==['src/module.py']
|
|
r=json.loads(s.collect(query('absent','src/module.py')))['result']
|
|
assert r['items']==[] and not r['truncated']
|
|
assert 'global absence' in r['interpretation']
|
|
|
|
|
|
def test_registration_name_does_not_expand_unrelated_loader_symbols(source):
|
|
r=json.loads(service(source).collect(query('AddSC_events')))['result']
|
|
assert [q['query']['text'] for q in r['searches']]==['AddSC_events']
|
|
assert not r['truncated']
|
|
|
|
|
|
def test_long_lines_and_missing_blocks_are_explicit(source):
|
|
r=json.loads(service(source).collect(query('LONG_TARGET')))['result']
|
|
assert 'line_text_limit' in r['limits_reached']
|
|
assert r['items'][0]['blocks'][0]['truncated_lines']==[1]
|
|
assert len(r['items'][0]['blocks'][0]['lines'][0])==2048
|
|
r=json.loads(service(source,context_lines=1,max_blocks=1).collect(query('GAPPED')))['result']
|
|
assert {'block_limit','omitted_anchors'} <= set(r['limits_reached'])
|
|
assert any(not a['context_returned'] for a in r['items'][0]['anchors'])
|
|
|
|
|
|
@pytest.mark.parametrize('prefix',['../outside','/etc','C:/outside','.git/','a\\b','src/\x00'])
|
|
def test_unsafe_paths_rejected_before_git(source,monkeypatch,prefix):
|
|
s=service(source)
|
|
monkeypatch.setattr(source,'query',lambda args:pytest.fail('must not access repository'))
|
|
with pytest.raises(ToolError):s.collect(query(prefix=prefix))
|
|
|
|
|
|
@pytest.mark.parametrize('text',['',' ','x\ny','x\x00y','x\ty'])
|
|
def test_invalid_text_is_controlled(source,text):
|
|
with pytest.raises(ToolError):service(source).collect(query(text))
|
|
|
|
|
|
def test_no_regex_shell_or_git_options(source):
|
|
s=service(source)
|
|
assert json.loads(s.collect(query('.*')))['result']['items']==[]
|
|
assert json.loads(s.collect(query('--help')))['result']['items']==[]
|
|
with pytest.raises(ToolError):s.collect({**query(),'command':'git fetch'})
|
|
with pytest.raises(ToolError):s.collect({**query(),'_extensions':['cpp']})
|
|
|
|
|
|
@pytest.mark.parametrize('settings',[{'max_searches':0},{'max_matches':True},{'max_files':1.0},
|
|
{'registration_prefixes':['.*']},{'extensions':[]},{'extensions':['cpp;bad']},{'unknown':1}])
|
|
def test_invalid_configuration(source,settings):
|
|
with pytest.raises(ToolError):service(source,**settings)
|
|
|
|
|
|
@pytest.mark.parametrize('settings,reason',[({'max_searches':1},'followup_search_limit'),
|
|
({'max_matches':1},'match_limit'),({'max_files':1},'file_limit'),
|
|
({'max_total_lines':1},'line_limit')])
|
|
def test_limits_are_reported_without_claiming_complete(source,settings,reason):
|
|
r=json.loads(service(source,**settings).collect(query()))['result']
|
|
assert r['truncated'] and reason in r['limits_reached']
|
|
assert r['coverage']=='partial_configured_source_context'
|
|
|
|
|
|
def test_nondefault_settings_binding_changes_and_disabled_tool_stays_hidden(source):
|
|
one=service(source);two=service(source,context_lines=2)
|
|
assert json.loads(one.collect(query()))['origin']['source_analysis_sha256']!=json.loads(two.collect(query()))['origin']['source_analysis_sha256']
|
|
disabled=ProjectResearch({},source=source,source_analysis={'enabled':False})
|
|
assert disabled.query_schema()==ProjectResearch.schema()
|
|
with pytest.raises(ToolError):disabled.collect(query())
|
|
|
|
|
|
def test_byte_and_time_limits_are_controlled(source,monkeypatch):
|
|
source.limits=ProjectLimits(max_result_bytes=100)
|
|
s=service(source)
|
|
with pytest.raises(ToolError,match='RESULT_LIMIT'):s.collect(query())
|
|
assert not s._captures
|
|
import viano_bug_analysis.tools.source_context as module
|
|
clock=iter([0,100])
|
|
monkeypatch.setattr(module.time,'monotonic',lambda:next(clock))
|
|
with pytest.raises(ToolError,match='TIMEOUT'):s.collect(query())
|
|
|
|
|
|
def test_source_context_strict_contract_and_old_replay(source):
|
|
from test_investigation_lean import context
|
|
from viano_bug_analysis.analysis.investigation_prompt import output_schema
|
|
c=context();c.pop('lean_session',None)
|
|
action={'kind':'tool','operation':'project.query','arguments':query()}
|
|
assert not Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
|
|
c['tool_schema']['x-project-bindings']['source_analysis']=service(source).source_context.catalog()
|
|
assert Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
|
|
|
|
|
|
def test_changed_source_analysis_cannot_execute_a_frozen_context(source):
|
|
from types import SimpleNamespace
|
|
from viano_bug_analysis.analysis import Investigator
|
|
s=service(source)
|
|
c={'tool_schema':{'x-project-bindings':{'source':source.binding(),'database':None,
|
|
'source_analysis':s.source_context.catalog()}}}
|
|
loop=Investigator.__new__(Investigator);loop.router=SimpleNamespace(project_research=s)
|
|
loop._check_project_binding(c)
|
|
s.source_context=service(source,context_lines=1).source_context
|
|
with pytest.raises(ToolError,match='BINDING_CHANGED'):loop._check_project_binding(c)
|
|
|
|
|
|
def test_budgeted_evidence_and_portable_hybrid_resume(source,tool_case,tmp_path):
|
|
from test_investigation_hybrid import HybridAdapter
|
|
from test_case_interpreter import gateway
|
|
from test_investigator import run,state
|
|
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 CheckpointStore,Workspace
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
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()})
|
|
router=InvestigationToolRouter(project_research=s)
|
|
loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),router=router)
|
|
first=run(loop,steps=3)
|
|
assert all(h['iterations']==1 for h in state(first)['hypotheses'].values())
|
|
assert 'source_analysis' in adapter.calls[0]['input']['tool_schema']['project']
|
|
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
|
|
measurement=measure_package(final.package)
|
|
assert measurement['unmatched_requests']==0 and measurement['all_exactly_reconstructed']
|
|
budget=BudgetManager(portable).snapshot('case-A','stable-package').state
|
|
assert all(a['status']=='settled' for h in budget['accounts'].values() for a in h['actions'].values())
|
|
with PackageReader(final.package) as reader:
|
|
paths=[p for p in reader.files if p.startswith('analysis/evidence/captures/')]
|
|
capture=next(json.loads(reader.read_bytes(p)) for p in paths if json.loads(reader.read_bytes(p))['query']['kind']=='source.context')
|
|
assert capture['result']['items'][0]['git_blob']
|
|
assert capture['origin']['commit']==source.commit
|
|
|
|
|
|
def test_real_pinned_source_context():
|
|
import os
|
|
config=os.environ.get('VBA_TEST_PROJECT_CONFIG')
|
|
if not config:pytest.skip('Opt-in pinned local source checkout')
|
|
s=ProjectResearch.load(config)
|
|
capture=json.loads(s.collect(query('enum UnitFlags','src/server/')))
|
|
assert capture['origin']['commit']==s.source.commit
|
|
assert capture['result']['items']
|
|
for item in capture['result']['items']:
|
|
text,metadata=s.source._read(item['path'])
|
|
assert item['git_blob']==metadata['git_blob'] and item['sha256']==metadata['sha256']
|
|
lines=text.splitlines()
|
|
for block in item['blocks']:
|
|
assert block['lines']==[line[:2048] for line in lines[block['start_line']-1:block['start_line']-1+len(block['lines'])]]
|