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'])]]