120 lines
5.7 KiB
Python
120 lines
5.7 KiB
Python
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import ChangeRequestBuilder, OutputValidator
|
|
from viano_bug_analysis.analysis.change_request_contracts import (
|
|
build_prompt, compact_handoff_context, scope_evidence, validate_request,
|
|
ChangeRequestError,
|
|
)
|
|
from viano_bug_analysis.package import PackageReader
|
|
from test_change_request_builder import diagnosed, build, RequestAdapter
|
|
from test_case_interpreter import gateway
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_handoff_outcomes import minimal
|
|
from test_investigator import CASE
|
|
|
|
|
|
def test_capture_scope_is_explicit_and_schema_similarity_is_not_provenance():
|
|
payload, context = minimal()
|
|
context['local_context'] = {'capture.json': {'kind': 'database'},
|
|
'creature-schema.json': {'kind': 'database'}}
|
|
context['evidence']['E000001']['content']['items'] = [
|
|
{'query': {'kind': 'database.rows', 'table': 'creature'},
|
|
'result': {'items': [], 'coverage': 'complete_filtered_rows'}}]
|
|
assert scope_evidence(context) == {'capture.json': ['E000001']}
|
|
payload['affected_scope'] = [{'kind': 'database', 'source_ref': 'capture.json',
|
|
'evidence_refs': ['E000001']}]
|
|
validate_request(payload, context)
|
|
payload['affected_scope'][0]['source_ref'] = 'creature-schema.json'
|
|
with pytest.raises(ChangeRequestError, match='SCOPE_EVIDENCE'):
|
|
validate_request(payload, context)
|
|
|
|
|
|
def test_compact_view_keeps_findings_observations_uncertainty_and_context_unchanged():
|
|
_, context = minimal()
|
|
context.update(case_id='case', package_id='package', analysis_revision=1,
|
|
diagnosis_ref='diagnosis.json', diagnosis_sha256='0'*64,
|
|
interpretation={'unknowns': ['Event membership not fully checked']},
|
|
policy={'rules': ['No invented data']},
|
|
case={'human_observations': ['Creatures absent'],
|
|
'data_quality': {'warnings': ['Snapshot only']},
|
|
'inputs': ['unused input'] * 10000,
|
|
'source_context': {'refs': ['unused catalog'] * 10000}})
|
|
context['evidence']['E000001']['record']['content_sha256'] = '1'*64
|
|
original = copy.deepcopy(context)
|
|
view = compact_handoff_context(context)
|
|
assert view['evidence']['E000001']['content'] == context['evidence']['E000001']['content']
|
|
assert view['diagnosis'] == context['diagnosis']
|
|
assert view['interpretation'] == context['interpretation']
|
|
assert view['case']['human_observations'] == ['Creatures absent']
|
|
assert view['case']['data_quality'] == {'warnings': ['Snapshot only']}
|
|
request = build_prompt(context, compact=True)
|
|
assert json.loads(request['messages'][1]['content'])['output_schema'] == request['output_schema']
|
|
assert len(json.dumps(request)) < len(json.dumps(build_prompt(context))) / 5
|
|
assert context == original
|
|
|
|
|
|
@pytest.mark.parametrize('first_version', ['change_request_v2', 'change_request_v3'])
|
|
def test_retry_gets_actionable_feedback_and_portable_replay(diagnosed, tmp_path, first_version):
|
|
import asyncio
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace
|
|
|
|
class Correcting(RequestAdapter):
|
|
async def generate(self, config, request, **kwargs):
|
|
response = await super().generate(config, request, **kwargs)
|
|
assert request['prompt_version'] == (first_version if self.calls == 1 else 'change_request_v9')
|
|
if self.calls == 1:
|
|
response.document['affected_scope'][0]['source_ref'] = 'wrong.cpp'
|
|
else:
|
|
feedback = self.contexts[-1]['validation_feedback']
|
|
assert feedback[0]['code'] == 'CHANGE_REQUEST_SCOPE'
|
|
assert feedback[0]['rejected_scope'][0]['source_ref'] == 'wrong.cpp'
|
|
assert self.contexts[-1]['local_context']['tool/code.cpp']['allowed_evidence_refs']
|
|
return response
|
|
|
|
store, _ = diagnosed
|
|
adapter = Correcting()
|
|
g = gateway(adapter)
|
|
builder = ChangeRequestBuilder(store, g)
|
|
original = builder._request
|
|
def first_request(reader, context, operation, **kwargs):
|
|
if not adapter.calls:
|
|
kwargs['version'] = first_version
|
|
return original(reader, context, operation, **kwargs)
|
|
builder._request = first_request
|
|
assert build(builder).status == 'REJECTED'
|
|
result = build(builder)
|
|
assert result.status == 'COMPLETED'
|
|
assert build(builder).reused and adapter.calls == 2
|
|
exported = asyncio.run(OutputValidator(store, g).publish(*CASE))
|
|
portable = CheckpointStore(Workspace(tmp_path / 'recipient'))
|
|
portable.intake(exported.exported)
|
|
assert asyncio.run(OutputValidator(portable, g).publish(*CASE)).status == 'OUTPUT_READY'
|
|
assert adapter.calls == 2
|
|
|
|
|
|
def test_existing_v2_call_stays_frozen_after_upgrade(diagnosed):
|
|
store, _ = diagnosed
|
|
adapter = RequestAdapter()
|
|
builder = ChangeRequestBuilder(store, gateway(adapter))
|
|
original_context = builder.context
|
|
def legacy_context(reader, policy, **kwargs):
|
|
kwargs['parameter_grounding'] = 1
|
|
return original_context(reader, policy, **kwargs)
|
|
builder.context = legacy_context
|
|
original = builder._request
|
|
builder._request = lambda reader, context, operation, **kw: original(
|
|
reader, context, operation, version='change_request_v2')
|
|
publish = builder.publish
|
|
builder.publish = lambda *a, **kw: (_ for _ in ()).throw(RuntimeError('crash'))
|
|
with pytest.raises(ChangeRequestError):
|
|
build(builder)
|
|
builder._request = original
|
|
builder.context = original_context
|
|
builder.publish = publish
|
|
assert build(builder).status == 'COMPLETED'
|
|
assert build(builder).reused and adapter.calls == 1
|