94 lines
4.8 KiB
Python
94 lines
4.8 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import DiagnosisBuilder, DiagnosisError
|
|
from viano_bug_analysis.analysis.diagnosis_builder import build_prompt, compact_diagnosis_context
|
|
from viano_bug_analysis.analysis.change_request_contracts import build_prompt as handoff_prompt
|
|
from viano_bug_analysis.package import PackageReader
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_tools import tool_case
|
|
from test_diagnosis_builder import investigated, FinalAdapter, build
|
|
|
|
|
|
def test_diagnosis_view_retains_findings_limits_questions_and_actual_row_values():
|
|
context = {'case': {'inputs': {'catalog': ['inventory'] * 10000},
|
|
'source_context': {'refs': ['source inventory'] * 10000, 'warnings': ['Incomplete scan']},
|
|
'database_context': {'refs': ['db inventory'] * 10000},
|
|
'runtime_facts': [{'id': 'F0001', 'text': 'Actual runtime observation'}],
|
|
'human_observations': ['Invisible creature'], 'entities': [{'id': 123}],
|
|
'data_quality': {'warnings': ['Snapshot only']}},
|
|
'budget': {'accounts': {'H1': {'remaining': 50}}, 'seen_evidence': ['bookkeeping']},
|
|
'investigation': {'hypotheses': {'H1': {'confidence': 'supported',
|
|
'questions': ['Dynamic path untested'], 'notes': ['Static capture is not exhaustive']}},
|
|
'early_stop': {'status': 'requested', 'deferred_hypotheses': ['H2']}},
|
|
'interpretation': {'unknowns': ['Visibility not tested']},
|
|
'evidence': {'E000001': {'content': {'coverage': 'complete_filtered_rows',
|
|
'items': [{'query': {'filters': [{'column': 'id', 'value': 123}]},
|
|
'result': {'truncated': False, 'items': [{'row': {'id': 123,
|
|
'path': 'a'*64, 'sha256': 'b'*64, 'map': 0}}]}}]}}}}
|
|
original = copy.deepcopy(context)
|
|
view = compact_diagnosis_context(context)
|
|
assert view['evidence'] == context['evidence']
|
|
assert view['investigation'] == context['investigation']
|
|
assert view['case']['runtime_facts'] == context['case']['runtime_facts']
|
|
assert view['case']['source_context']['warnings'] == ['Incomplete scan']
|
|
assert view['case']['human_observations'] == ['Invisible creature']
|
|
assert view['interpretation'] == context['interpretation']
|
|
assert 'inputs' not in view['case'] and 'seen_evidence' not in view['budget']
|
|
assert len(json.dumps(view)) < len(json.dumps(context)) / 10
|
|
assert context == original
|
|
|
|
|
|
def test_existing_v2_diagnosis_request_survives_crash_and_upgrade(investigated):
|
|
store, _ = investigated
|
|
adapter = FinalAdapter(True)
|
|
builder = DiagnosisBuilder(store, gateway(adapter))
|
|
original = builder._request
|
|
def old_request(context, reader, attempt, **kwargs):
|
|
kwargs['compact'] = False
|
|
return original(context, reader, attempt, **kwargs)
|
|
builder._request = old_request
|
|
publish = builder._publish
|
|
def crash(*a, **kw):
|
|
raise OSError('simulated interruption')
|
|
builder._publish = crash
|
|
with pytest.raises(DiagnosisError):
|
|
build(builder)
|
|
assert adapter.requests[0]['prompt_version'] == 'diagnosis_v2'
|
|
builder._request = original
|
|
builder._publish = publish
|
|
assert build(builder).status == 'COMPLETED'
|
|
assert build(builder).reused and adapter.calls == 1
|
|
|
|
|
|
def test_new_diagnosis_keeps_frozen_context_and_uses_compact_request(investigated):
|
|
store, _ = investigated
|
|
adapter = FinalAdapter(True)
|
|
result = build(DiagnosisBuilder(store, gateway(adapter)))
|
|
request = adapter.requests[0]
|
|
assert request['prompt_version'] == 'diagnosis_v8'
|
|
visible = restore(json.loads(request['messages'][1]['content'])['input'])
|
|
with PackageReader(result.package) as reader:
|
|
doc = json.loads(reader.read_bytes(result.diagnosis_ref))
|
|
context = json.loads(reader.read_bytes(doc['context_ref']))
|
|
assert 'inputs' in context['case'] and 'inputs' not in visible['case']
|
|
assert request['output_schema'] == build_prompt(context, modern=True, lean=True)['output_schema']
|
|
|
|
|
|
def test_handoff_v4_accepts_reviewed_remediation_data_without_rewriting_old_prompt():
|
|
from test_handoff_outcomes import minimal
|
|
_, context = minimal()
|
|
context.update(case_id='case', package_id='package', analysis_revision=1,
|
|
diagnosis_ref='diagnosis.json', diagnosis_sha256='0'*64,
|
|
interpretation={}, policy={}, case={})
|
|
context['evidence']['E000001']['record']['content_sha256'] = '1'*64
|
|
old = handoff_prompt(context, compact=True)
|
|
new = handoff_prompt(context, compact=True, version='change_request_v4')
|
|
assert old['prompt_version'] == 'change_request_v3'
|
|
assert 'Human-selected and reviewed placements are acceptable' in new['messages'][0]['content']
|
|
assert new['output_schema'] == old['output_schema']
|