127 lines
5.8 KiB
Python
127 lines
5.8 KiB
Python
import copy
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from test_project_research import source
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_bounded_contract_v2 import loop_for
|
|
from test_investigation_startup import Adapter
|
|
from test_investigator import run, state
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch, ProjectLimits
|
|
from viano_bug_analysis.package import PackageValidator
|
|
|
|
FILE = 'src/server/scripts/EasternKingdoms/arathi_highlands.cpp'
|
|
|
|
|
|
@pytest.fixture
|
|
def long_source(source):
|
|
path = source.path / FILE
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text(''.join(f'code line {i}\n' for i in range(1, 301)), encoding='utf-8')
|
|
def git(*args):
|
|
return subprocess.run(['git', '-C', str(source.path), *args], check=True,
|
|
capture_output=True).stdout.decode().strip()
|
|
git('add', FILE)
|
|
git('-c', 'user.name=Test', '-c', 'user.email=test@example.invalid', 'commit', '-m', 'long fixture')
|
|
source.commit = git('rev-parse', 'HEAD')
|
|
source._inventory = None
|
|
return source
|
|
|
|
|
|
def read(start=115, end=215):
|
|
return {'kind':'source.read', 'path':FILE, 'start_line':start, 'end_line':end}
|
|
|
|
|
|
def test_clamp_is_explicit_and_next_page_recovers_requested_lines(long_source):
|
|
query = read()
|
|
before = copy.deepcopy(query)
|
|
result = long_source.query(query)
|
|
assert query == before
|
|
assert len(result['items']) == 80
|
|
assert result['requested_line_count'] == 101 and result['line_limit'] == 80
|
|
assert result['requested_start_line'] == 115 and result['requested_end_line'] == 215
|
|
assert result['returned_start_line'] == 115 and result['returned_end_line'] == 194
|
|
assert result['truncated'] and not result['requested_range_complete'] and not result['eof']
|
|
assert result['next_start_line'] == 195
|
|
assert '101' in result['read_notice'] and '80' in result['read_notice']
|
|
following = long_source.query(read(result['next_start_line'], query['end_line']))
|
|
assert not following['truncated']
|
|
assert [v['line'] for v in result['items'] + following['items']] == list(range(115,216))
|
|
|
|
|
|
@pytest.mark.parametrize('start,end,count,truncated,eof,next_line', [
|
|
(1,80,80,False,False,None), (1,81,80,True,False,81),
|
|
(221,1000,80,False,True,None), (290,1000,11,False,True,None),
|
|
(301,1000,0,False,True,None),
|
|
])
|
|
def test_boundaries_and_eof(long_source,start,end,count,truncated,eof,next_line):
|
|
result = long_source.query(read(start,end))
|
|
assert len(result['items']) == count and result['truncated'] == truncated
|
|
assert result['eof'] == eof and result.get('next_start_line') == next_line
|
|
if end-start+1 > 80:
|
|
assert result['requested_range_complete'] == (not truncated)
|
|
if not count:
|
|
assert result['returned_start_line'] is None and result['returned_end_line'] is None
|
|
|
|
|
|
@pytest.mark.parametrize('start,end', [(0,80),(-1,80),(50,49),(True,80),(1,80.0)])
|
|
def test_invalid_ranges_still_rejected(long_source,start,end):
|
|
with pytest.raises(ToolError,match='PROJECT_SOURCE_RANGE'):
|
|
long_source.query(read(start,end))
|
|
|
|
|
|
def test_configurable_limit_and_unsafe_path(long_source):
|
|
long_source.limits = ProjectLimits(max_lines=7)
|
|
result = long_source.query(read(2,99))
|
|
assert len(result['items']) == 7 and result['next_start_line'] == 9
|
|
with pytest.raises(ToolError):
|
|
long_source.query({**read(), 'path':'../outside.cpp'})
|
|
|
|
|
|
@pytest.mark.parametrize('name', ['range101','range91'])
|
|
def test_actual_bug17_rejections_now_produce_frozen_partial_capture(long_source,name):
|
|
corpus = Path(__file__).parent / 'fixtures/llm_rejections/20261020/bug17-source-range' / name
|
|
answer = json.loads((corpus/'parsed-answer.json').read_text(encoding='utf-8'))
|
|
query = answer['next']['arguments']
|
|
before = copy.deepcopy(query)
|
|
service = ProjectResearch({}, source=long_source)
|
|
capture = json.loads(service.collect(query))
|
|
assert capture['query'] == query == before
|
|
assert capture['result']['truncated'] and len(capture['result']['items']) == 80
|
|
assert service.collect(query) == service.collect(query)
|
|
|
|
|
|
def test_partial_scope_survives_bounded_evidence_projection(long_source):
|
|
from viano_bug_analysis.analysis.investigation_window import evidence_page
|
|
capture = json.loads(ProjectResearch({},source=long_source).collect(read()))
|
|
page = evidence_page({'record':{'hypothesis_id':'H1'},'content':{'items':[capture]}})
|
|
assert page['encoding'] == 'json_fragment' and not page['complete_payload']
|
|
scope = page['scope'][0]
|
|
assert scope['truncated'] and not scope['requested_range_complete']
|
|
assert scope['returned_end_line'] == 194 and scope['next_start_line'] == 195
|
|
assert scope['line_limit'] == 80 and scope['requested_line_count'] == 101
|
|
|
|
|
|
def test_oversized_read_runs_in_controller_without_invalid_retry(tool_case,long_source):
|
|
class Read(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body = json.loads(request['messages'][1]['content'])
|
|
if body['task'] != 'startup_choose_query':
|
|
return await super().generate(config,request,**kwargs)
|
|
self.calls.append(body)
|
|
return ProviderResponse({'assessment':None,'observation':'Read a bounded source excerpt.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':read()}},'fixture',100,50)
|
|
adapter = Read()
|
|
result = run(loop_for(tool_case,adapter,source=long_source),steps=3)
|
|
saved = state(result)
|
|
assert saved['startup']['explored'] == ['H1','H2','H3']
|
|
assert len(adapter.calls) == 3
|
|
assert all(h['invalid_responses'] == 0 for h in saved['hypotheses'].values())
|
|
assert PackageValidator().validate(result.package).valid
|