137 lines
6.6 KiB
Python
137 lines
6.6 KiB
Python
import copy
|
|
|
|
from viano_bug_analysis.analysis.investigation_consolidation import consolidated_view, restore_view
|
|
from viano_bug_analysis.analysis.investigation_view import working_view, thin_view
|
|
from viano_bug_analysis.analysis.model_context import model_json
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from test_investigation_view import context
|
|
|
|
|
|
def capture(lines, *, sha='a'*64, commit='b'*40):
|
|
return {'origin': {'kind': 'git', 'commit': commit, 'url': 'https://example.invalid/repo'},
|
|
'query': {'kind': 'source.read', 'path': 'code.cpp', 'start_line': min(lines), 'end_line': max(lines)},
|
|
'result': {'coverage': 'selected_lines', 'truncated': False,
|
|
'items': [{'path': 'code.cpp', 'sha256': sha, 'git_blob': sha[:40], 'size': 10000,
|
|
'line': i, 'text': 'code line '+str(i), 'text_truncated': False} for i in lines]}}
|
|
|
|
|
|
def populated():
|
|
c = context()
|
|
c['prompt_version'] = 'investigation_project_v6'
|
|
c['evidence']['E000001']['content']['items'] = [capture(range(1, 81))]
|
|
c['evidence']['E000002']['content']['items'] = [capture(range(41, 121))]
|
|
return c
|
|
|
|
|
|
def roundtrip(c):
|
|
original = copy.deepcopy(c)
|
|
view = consolidated_view(c, require_savings=False)
|
|
assert model_json(restore_view(view)) == model_json(working_view(c))
|
|
assert c == original
|
|
assert model_json(consolidated_view(c, require_savings=False)) == model_json(view)
|
|
assert model_json(restore_view(consolidated_view(c))) == model_json(working_view(c))
|
|
assert len(model_json(consolidated_view(c))) <= len(model_json(thin_view(c)))
|
|
return view
|
|
|
|
|
|
def test_overlapping_lines_share_one_versioned_source_without_losing_scope():
|
|
c = populated()
|
|
v = roundtrip(c)
|
|
assert len(v['shared_sources']) == 1
|
|
source = v['shared_sources']['FILE0001']
|
|
assert len(source['lines']['rows']) == 120
|
|
assert source['lines']['common'] == {'text_truncated': False}
|
|
assert source['conflicting_lines'] == []
|
|
assert len(model_json(v)) < len(model_json(thin_view(c)))
|
|
for eid, bounds in (('E000001', (1, 80)), ('E000002', (41, 120))):
|
|
q = v['evidence'][eid]['content']['items'][0]['query']
|
|
assert q['path_ref'] == 'FILE0001' and 'path' not in q
|
|
assert (q['start_line'], q['end_line']) == bounds
|
|
|
|
|
|
def test_gaps_duplicates_order_and_conflicts_are_preserved():
|
|
c = populated()
|
|
c['evidence']['E000001']['content']['items'] = [capture([1, 3, 3])]
|
|
second = capture([3, 5])
|
|
second['result']['items'][0]['text'] = 'contradictory observation'
|
|
c['evidence']['E000002']['content']['items'] = [second]
|
|
source = roundtrip(c)['shared_sources']['FILE0001']
|
|
assert source['conflicting_lines'] == [3]
|
|
rows = source['lines']['rows']
|
|
assert len(rows) == 4 # duplicate line 3 shared; conflicting line 3 retained.
|
|
assert not any(row[0] in (2, 4) for row in rows)
|
|
|
|
|
|
def test_revision_and_snapshot_differences_do_not_merge_sources():
|
|
c = populated()
|
|
c['evidence']['E000002']['content']['items'] = [capture(range(41, 121), sha='c'*64)]
|
|
assert len(roundtrip(c)['shared_sources']) == 2
|
|
c['evidence']['E000002']['content']['items'] = [capture(range(41, 121), commit='d'*40)]
|
|
assert len(roundtrip(c)['shared_sources']) == 2
|
|
|
|
|
|
def test_missing_origin_is_not_assumed_to_be_the_same_snapshot():
|
|
c = populated()
|
|
del c['evidence']['E000002']['content']['items'][0]['origin']
|
|
view = roundtrip(c)
|
|
assert len(view['shared_sources']) == 1
|
|
q = view['evidence']['E000002']['content']['items'][0]['query']
|
|
assert q['path'] == 'code.cpp' and 'path_ref' not in q
|
|
|
|
|
|
def test_identical_database_results_keep_separate_filters_origins_and_types():
|
|
c = populated()
|
|
result = {'coverage': 'partial_filtered_rows', 'truncated': True, 'next_offset': 2,
|
|
'items': [{'row': {'name': 'a'*200, 'value': None, 'enabled': False}},
|
|
{'row': {'name': 'b'*200, 'value': 0, 'enabled': 1}}]}
|
|
for i, entry in enumerate(c['evidence'].values()):
|
|
entry['content']['items'] = [{'query': {'kind': 'database.rows', 'table': 'example', 'filters': [i]},
|
|
'origin': {'snapshot_id': str(i)}, 'result': copy.deepcopy(result)}]
|
|
view = roundtrip(c)
|
|
assert len(view['shared_results']) == 1
|
|
assert view['evidence']['E000001']['content']['items'][0]['query']['filters'] == [0]
|
|
assert view['evidence']['E000002']['content']['items'][0]['query']['filters'] == [1]
|
|
|
|
|
|
def test_old_v5_requests_stay_frozen_and_new_schema_is_unchanged():
|
|
c = populated()
|
|
c['prompt_version'] = 'investigation_project_v5'
|
|
old = build_prompt(c)
|
|
assert old['messages'][1]['content'] == model_json({'task': 'investigate_hypothesis', 'input': thin_view(c), 'output_schema': old['output_schema']})
|
|
c['prompt_version'] = 'investigation_project_v6'
|
|
new = build_prompt(c)
|
|
assert new['output_schema'] == old['output_schema']
|
|
assert new['messages'][1]['content'] == model_json({'task': 'investigate_hypothesis', 'input': consolidated_view(c), 'output_schema': new['output_schema']})
|
|
|
|
|
|
def test_heavily_repeated_code_is_transmitted_once():
|
|
c = populated()
|
|
for i in range(3, 13):
|
|
entry = copy.deepcopy(c['evidence']['E000001'])
|
|
entry['record']['id'] = f'E{i:06d}'
|
|
c['evidence'][f'E{i:06d}'] = entry
|
|
v = roundtrip(c)
|
|
assert len(v['shared_sources']) == 1
|
|
assert len(v['shared_sources']['FILE0001']['lines']['rows']) == 120
|
|
assert len(model_json(consolidated_view(c))) < len(model_json(thin_view(c))) / 2
|
|
|
|
|
|
def test_shared_source_paths_resolve_only_inside_the_active_context(monkeypatch):
|
|
from types import SimpleNamespace
|
|
from viano_bug_analysis.analysis.investigator import Investigator
|
|
c = populated()
|
|
state = {'case_id':'case','package_id':'package','analysis_revision':1,'prompt_protocol':2,
|
|
'active':{'hypothesis_id':'H1','iteration':1,'context_ref':'context.json'}}
|
|
class Reader:
|
|
def __init__(self, *args): pass
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
monkeypatch.setattr('viano_bug_analysis.analysis.investigator.PackageReader', Reader)
|
|
fake = SimpleNamespace(store=SimpleNamespace(limits=None), _latest=lambda *a:'package.zip', _read=lambda *a:c)
|
|
for path, expected in [('FILE0001', 'code.cpp'), ('FILE9999', 'FILE9999'), ('other.cpp', 'other.cpp')]:
|
|
action = {'operation':'project.query','arguments':{'kind':'source.read','path':path,'start_line':1,'end_line':3}}
|
|
original = copy.deepcopy(action)
|
|
request = Investigator._tool_request(fake, state, action)
|
|
assert request['requests'][0]['arguments']['path'] == expected
|
|
assert action == original
|