182 lines
9.4 KiB
Python
182 lines
9.4 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import DiagnosisBuilder, DiagnosisError, Investigator, InvestigationError
|
|
from viano_bug_analysis.analysis.early_stop import check_request, evidence_fingerprint
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.lifecycle import State, LifecycleError
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_tools import tool_case
|
|
from test_investigator import CASE, LoopAdapter, run, state
|
|
from test_diagnosis_builder import FinalAdapter, build
|
|
|
|
|
|
def request_early(payload, context):
|
|
early = context['early_review']['request']
|
|
if early and early['status'] == 'preparing':
|
|
payload['next'] = {'kind': 'defer', 'reason': 'Assess pending result before independent review'}
|
|
elif context['selected_hypothesis'] == 'H1' and context['previous_result']:
|
|
payload['update']['confidence'] = 'strongly_supported'
|
|
other = next(eid for eid, e in context['evidence'].items() if e['record']['hypothesis_id'] == 'H2')
|
|
payload['next'] = {'kind': 'request_diagnosis', 'reason': 'Direct evidence and targeted alternative check',
|
|
'countercheck_refs': [other]}
|
|
|
|
|
|
@pytest.fixture
|
|
def early_case(tool_case):
|
|
store = tool_case[0]
|
|
adapter = LoopAdapter(change=request_early)
|
|
loop = Investigator(store, gateway(adapter))
|
|
result = run(loop)
|
|
assert result.status == 'COMPLETED'
|
|
saved = state(result)
|
|
assert saved['early_stop']['status'] == 'requested'
|
|
assert saved['early_stop']['deferred_hypotheses'] == ['H2', 'H3']
|
|
assert all(h['disposition'] == 'active' for h in saved['hypotheses'].values())
|
|
budget = BudgetManager(store).snapshot(*CASE)
|
|
assert not budget.ready_for_evaluation
|
|
assert all(a['status'] == 'settled' for account in budget.state['accounts'].values() for a in account['actions'].values())
|
|
assert len(adapter.contexts) == 6
|
|
return store, result.package, budget.state
|
|
|
|
|
|
def test_accept_early_with_remaining_budgets_and_replay(early_case):
|
|
store, original, budget = early_case
|
|
adapter = FinalAdapter(True, lambda p, c: p.update(confidence='strongly_supported'))
|
|
builder = DiagnosisBuilder(store, gateway(adapter))
|
|
result = build(builder)
|
|
assert result.status == 'COMPLETED'
|
|
assert build(builder).reused and adapter.calls == 1
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(original) as old, PackageReader(result.package) as new:
|
|
assert new.manifest['artifacts']['change_request'] is None
|
|
for name in old.files:
|
|
if name != 'manifest.json':
|
|
assert old.read_bytes(name) == new.read_bytes(name)
|
|
from viano_bug_analysis.analysis import ChangeRequestBuilder, OutputValidator
|
|
from test_change_request_builder import RequestAdapter
|
|
request_adapter = RequestAdapter()
|
|
request_gateway = gateway(request_adapter)
|
|
assert build(ChangeRequestBuilder(store, request_gateway)).status == 'COMPLETED'
|
|
exported = asyncio.run(OutputValidator(store, request_gateway).publish(*CASE))
|
|
assert exported.status == 'OUTPUT_READY' and exported.exported.exists()
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
assert request_adapter.calls == 1
|
|
|
|
|
|
@pytest.mark.parametrize('confirmed', [False, True])
|
|
def test_decline_resumes_without_spending_or_publishing(early_case, confirmed):
|
|
store, original, budget = early_case
|
|
# Merely supported is insufficient for early acceptance, even if confirmed.
|
|
result = build(DiagnosisBuilder(store, gateway(FinalAdapter(confirmed))))
|
|
assert result.status == 'CONTINUED'
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
with PackageReader(result.package) as reader:
|
|
assert reader.manifest['state'] == State.INVESTIGATING_HYPOTHESES
|
|
assert not any(reader.manifest['artifacts'].get(f) for f in ('diagnosis', 'failure_report', 'change_request', 'diagnosis_context'))
|
|
saved = Investigator(store, gateway(LoopAdapter()))._load(reader)
|
|
assert saved['early_stop']['status'] == 'declined'
|
|
assert not saved['completed'] and saved['summary_ref'] is None
|
|
assert reader.read_bytes(saved['early_stop']['review_ref'])
|
|
# Continue without reassessing or charging the same results twice, then take
|
|
# the ordinary diagnosis path, whose namespace must remain unused.
|
|
adapter = LoopAdapter(change=lambda p,c: p.update(next={'kind':'finish','reason':'Remaining questions assessed'}))
|
|
done = run(Investigator(store, gateway(adapter)))
|
|
assert done.status == 'COMPLETED'
|
|
assert all(c['previous_result'] is None for c in adapter.contexts)
|
|
assert BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['remaining'] == budget['accounts']['H1']['remaining']
|
|
assert build(DiagnosisBuilder(store, gateway(FinalAdapter(True)))).status == 'COMPLETED'
|
|
|
|
|
|
def test_ordinary_diagnosis_cannot_rewind(tool_case):
|
|
store = tool_case[0]
|
|
run(Investigator(store, gateway(LoopAdapter())))
|
|
with pytest.raises(LifecycleError):
|
|
store.advance(*CASE, State.INVESTIGATING_HYPOTHESES, step_id='illegal-rewind', details={'early_diagnosis_declined': True})
|
|
|
|
|
|
def test_second_request_keeps_both_contexts(early_case):
|
|
store, _, _ = early_case
|
|
build(DiagnosisBuilder(store, gateway(FinalAdapter())))
|
|
def new_evidence(payload, context):
|
|
if context['selected_hypothesis'] == 'H1' and context['previous_result'] is None:
|
|
payload['next']['arguments'].update(start_line=3, end_line=4)
|
|
elif context['selected_hypothesis'] != 'H1':
|
|
payload['next'] = {'kind': 'finish', 'reason': 'Alternative adequately checked'}
|
|
request_early(payload, context)
|
|
result = run(Investigator(store, gateway(LoopAdapter(change=new_evidence))))
|
|
assert state(result)['early_stop']['number'] == 2
|
|
builder = DiagnosisBuilder(store, gateway(FinalAdapter(True, lambda p,c: p.update(confidence='strongly_supported'))))
|
|
assert build(builder).status == 'COMPLETED'
|
|
assert build(builder).reused
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert reader.read_bytes('analysis/diagnosis/0001/early/0001/context.json')
|
|
assert reader.read_bytes('analysis/diagnosis/0001/early/0001/review.json')
|
|
assert reader.read_bytes('analysis/diagnosis/0001/early/0002/context.json')
|
|
|
|
|
|
def test_decline_replays_model_call_after_interruption(early_case, monkeypatch):
|
|
store, _, budget = early_case
|
|
adapter = FinalAdapter()
|
|
builder = DiagnosisBuilder(store, gateway(adapter))
|
|
original = builder._decline
|
|
def crash(*args):
|
|
raise RuntimeError('interrupted after model result')
|
|
monkeypatch.setattr(builder, '_decline', crash)
|
|
with pytest.raises(DiagnosisError, match="DIAGNOSIS_INPUT"):
|
|
build(builder)
|
|
monkeypatch.setattr(builder, '_decline', original)
|
|
assert build(builder).status == 'CONTINUED' and adapter.calls == 1
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
|
|
|
|
def test_invalid_review_exhaustion_resumes_investigation(early_case):
|
|
from viano_bug_analysis.analysis import DiagnosisConfig
|
|
store, _, budget = early_case
|
|
adapter = FinalAdapter(True, lambda p,c: p.update(selected_hypothesis='H999'))
|
|
result = build(DiagnosisBuilder(store, gateway(adapter), DiagnosisConfig(1)))
|
|
assert result.status == 'CONTINUED'
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
|
|
|
|
def gate_fixture():
|
|
evidence = {f'E{i:06d}': {'record': {'hypothesis_id': f'H{i}', 'content_sha256': str(i)}, 'content': {'items': [i]}} for i in (1, 2)}
|
|
context = {'selected_hypothesis': 'H1', 'evidence': evidence}
|
|
saved = {'policy': {'max_early_reviews': 3}, 'hypotheses': {'H1': {'confidence': 'strongly_supported'}, 'H2': {'confidence': 'weak'}}}
|
|
decision = {'update': {'confidence': 'strongly_supported', 'evidence_refs': ['E000001']}, 'next': {'countercheck_refs': ['E000002']}}
|
|
budget = {'accounts': {'H1': {'actions': {}}, 'H2': {'actions': {}}}}
|
|
return saved, context, decision, budget
|
|
|
|
|
|
@pytest.mark.parametrize('change,code', [
|
|
(lambda s,c,d,b: d['update'].update(confidence='supported'), 'STRONG_EVIDENCE'),
|
|
(lambda s,c,d,b: d['next'].update(countercheck_refs=[]), 'COUNTERCHECK'),
|
|
(lambda s,c,d,b: s.update(early_stop_attempts=3), 'LIMIT'),
|
|
(lambda s,c,d,b: s.update(early_stop={'status':'declined','fingerprint':evidence_fingerprint(c['evidence'])}), 'NEW_EVIDENCE'),
|
|
(lambda s,c,d,b: b['accounts']['H1']['actions'].update(A={'status':'awaiting_assessment'}), 'PENDING_RESULT'),
|
|
])
|
|
def test_request_gates(change, code):
|
|
args = gate_fixture()
|
|
check_request(*args)
|
|
change(*args)
|
|
with pytest.raises(InvestigationError, match=code):
|
|
check_request(*args)
|
|
|
|
|
|
def test_duplicate_content_is_not_new_evidence():
|
|
_, context, _, _ = gate_fixture()
|
|
before = evidence_fingerprint(context['evidence'])
|
|
context['evidence']['E000003'] = context['evidence']['E000001']
|
|
assert evidence_fingerprint(context['evidence']) == before
|
|
|
|
|
|
def test_review_status_and_codes_remain_observable():
|
|
from viano_bug_analysis.telemetry.safety import STATUSES, safe_code
|
|
assert 'CONTINUED' in STATUSES
|
|
for code in ('EARLY_DIAGNOSIS_DECLINED', 'EARLY_STOP_COUNTERCHECK_REQUIRED', 'EARLY_STOP_NEW_EVIDENCE_REQUIRED'):
|
|
assert safe_code(code) == code
|