231 lines
12 KiB
Python
231 lines
12 KiB
Python
import copy
|
|
import json
|
|
from importlib.resources import files
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.budget import BudgetManager, BudgetError
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.evidence import EvidenceStore, EvidenceError, EvidenceLimits
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace, State
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from test_investigation_tools import tool_case, batch
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case
|
|
from test_hypothesis_builder import interpreted_case
|
|
from test_budget_manager import hypotheses
|
|
from test_async_tests import setup as async_setup
|
|
|
|
CASE=("case-A","stable-package")
|
|
|
|
|
|
def prepare(tool_case, *, action="A1", text="activate", budget=None):
|
|
store=tool_case[0]
|
|
budget=budget or BudgetManager(store)
|
|
budget.initialize(*CASE)
|
|
reserved=budget.reserve(*CASE,"H1",action,"source.find_text")
|
|
request=batch()
|
|
request["requests"][0].update(action_id=action,request_id=action)
|
|
request["requests"][0]["arguments"]["text"]=text
|
|
result=InvestigationToolRouter().query(reserved.package,request)
|
|
return budget,request,result
|
|
|
|
|
|
def test_roundtrip_stable_ids_provenance_preservation_and_budget(tool_case,tmp_path):
|
|
budget,request,result=prepare(tool_case)
|
|
before=budget.snapshot(*CASE)
|
|
evidence=EvidenceStore(tool_case[0])
|
|
saved=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
assert PackageValidator().validate(saved.package).valid
|
|
assert budget.snapshot(*CASE).state==before.state
|
|
with PackageReader(before.package) as old, PackageReader(saved.package) as reader:
|
|
record=json.loads(reader.read_bytes(saved.record_refs[0]))
|
|
content=json.loads(reader.read_bytes(record["content_ref"]))
|
|
assert record["id"]=="E000001" and record["duplicate_of"] is None
|
|
assert len(content["items"])==3
|
|
assert record["origin"]["position"]==0
|
|
assert reader.manifest["package_generation"]==old.manifest["package_generation"]+1
|
|
for name in old.files:
|
|
if name!="manifest.json": assert old.read_bytes(name)==reader.read_bytes(name)
|
|
assert evidence.store_tool_results(*CASE,"query1",request,result).reused
|
|
assert CaseBuilder(tool_case[0]).build(*CASE).reused
|
|
budget.record_evidence(*CASE,"H1","A1",saved.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","A1",semantic_score=100,reason="Relevant excerpt")
|
|
assert settled.state["accounts"]["H1"]["remaining"]==98
|
|
portable=CheckpointStore(Workspace(tmp_path/"portable"))
|
|
portable.intake(settled.package)
|
|
restored=EvidenceStore(portable).snapshot(*CASE)
|
|
assert restored.record_refs==saved.record_refs
|
|
assert BudgetManager(portable).snapshot(*CASE).state==settled.state
|
|
|
|
|
|
def test_metadata_changes_do_not_make_duplicates_novel(tool_case):
|
|
budget,request,result=prepare(tool_case)
|
|
evidence=EvidenceStore(tool_case[0])
|
|
first=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
budget.record_evidence(*CASE,"H1","A1",first.record_refs[0])
|
|
budget.settle(*CASE,"H1","A1",semantic_score=100,reason="Relevant")
|
|
budget,request,result=prepare(tool_case,action="A2",budget=budget)
|
|
second=evidence.store_tool_results(*CASE,"query2",request,result)
|
|
with PackageReader(second.package) as reader:
|
|
one=json.loads(reader.read_bytes(first.record_refs[0]))
|
|
two=json.loads(reader.read_bytes(second.record_refs[0]))
|
|
assert two["id"]=="E000002" and two["duplicate_of"]==one["id"]
|
|
assert two["content_ref"]==one["content_ref"]
|
|
budget.record_evidence(*CASE,"H1","A2",second.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","A2",semantic_score=100,reason="Duplicate")
|
|
assert settled.state["accounts"]["H1"]["actions"]["A2"]["reward"]==0
|
|
|
|
|
|
def test_empty_results_remain_auditable_without_novelty_reward(tool_case):
|
|
budget,request,result=prepare(tool_case,text="not-present")
|
|
saved=EvidenceStore(tool_case[0]).store_tool_results(*CASE,"empty",request,result)
|
|
budget.record_evidence(*CASE,"H1","A1",saved.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","A1",semantic_score=100,reason="Scoped absence")
|
|
action=settled.state["accounts"]["H1"]["actions"]["A1"]
|
|
assert action["novelty_score"]==action["reward"]==0
|
|
|
|
|
|
def test_overlapping_excerpts_only_reward_new_items(tool_case):
|
|
store=tool_case[0]; budget=BudgetManager(store); budget.initialize(*CASE)
|
|
first=budget.reserve(*CASE,"H1","A1","source.find_text")
|
|
request=batch();request["requests"][0]["arguments"]["limit"]=1
|
|
result=InvestigationToolRouter().query(first.package,request)
|
|
evidence=EvidenceStore(store)
|
|
saved=evidence.store_tool_results(*CASE,"one-line",request,result)
|
|
budget.record_evidence(*CASE,"H1","A1",saved.record_refs[0])
|
|
budget.settle(*CASE,"H1","A1",semantic_score=100,reason="First line")
|
|
second=budget.reserve(*CASE,"H1","A2","source.read_context")
|
|
request=batch("source.read_context",{"source_ref":"tool/code.cpp","start_line":1,"end_line":2})
|
|
request["requests"][0].update(action_id="A2")
|
|
result=InvestigationToolRouter().query(second.package,request)
|
|
saved=evidence.store_tool_results(*CASE,"overlap",request,result)
|
|
with PackageReader(saved.package) as reader:
|
|
record=json.loads(reader.read_bytes(saved.record_refs[0]))
|
|
assert record["novelty_score"]==50 and len(record["novel_atoms"])==1
|
|
budget.record_evidence(*CASE,"H1","A2",saved.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","A2",semantic_score=100,reason="Only one new line")
|
|
assert settled.state["accounts"]["H1"]["actions"]["A2"]["reward"]==4
|
|
|
|
|
|
def test_history_readable_after_reopen_and_ids_not_reset(tool_case):
|
|
import asyncio
|
|
from viano_bug_analysis.analysis import CaseInterpreter,HypothesisBuilder
|
|
from test_case_interpreter import gateway,FixtureAdapter
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
budget,request,result=prepare(tool_case)
|
|
store=tool_case[0];evidence=EvidenceStore(store)
|
|
first=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
budget.record_evidence(*CASE,"H1","A1",first.record_refs[0])
|
|
budget.settle(*CASE,"H1","A1",semantic_score=100,reason="Known excerpt")
|
|
store.advance(*CASE,State.FAILED_TO_ANALYZE,step_id="review",reason="Review")
|
|
store.advance(*CASE,State.REOPENED,step_id="reopen")
|
|
assert evidence.snapshot(*CASE).record_refs==first.record_refs
|
|
CaseBuilder(store).build(*CASE)
|
|
asyncio.run(CaseInterpreter(store,gateway(FixtureAdapter())).interpret(*CASE))
|
|
asyncio.run(HypothesisBuilder(store,gateway(ProposalAdapter())).build(*CASE))
|
|
budget=BudgetManager(store);budget.initialize(*CASE)
|
|
reserved=budget.reserve(*CASE,"H1","A1","source.find_text")
|
|
request=batch();request["analysis_revision"]=2
|
|
result=InvestigationToolRouter().query(reserved.package,request)
|
|
second=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
with PackageReader(second.package) as reader:
|
|
record=json.loads(reader.read_bytes(second.record_refs[0]))
|
|
assert record["id"]=="E000002" and record["duplicate_of"]=="E000001"
|
|
budget.record_evidence(*CASE,"H1","A1",second.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","A1",semantic_score=100,reason="Old evidence")
|
|
assert settled.state["accounts"]["H1"]["rewarded"]==0
|
|
|
|
|
|
@pytest.mark.parametrize("change",[
|
|
lambda r:r.update(package_generation=0),
|
|
lambda r:r["results"][0]["data"]["matches"][0].update(text="Invented code"),
|
|
lambda r:r["results"][0].update(hypothesis_id="H2"),
|
|
lambda r:r.update(batch_sha256="0"*64),
|
|
])
|
|
def test_forged_results_rejected_atomically(tool_case,change):
|
|
budget,request,result=prepare(tool_case)
|
|
before=budget.snapshot(*CASE)
|
|
change(result)
|
|
with pytest.raises(EvidenceError):
|
|
EvidenceStore(tool_case[0]).store_tool_results(*CASE,"bad",request,result)
|
|
assert budget.snapshot(*CASE).package==before.package
|
|
|
|
|
|
def test_conflicting_operation_and_missing_reservation(tool_case):
|
|
store=tool_case[0]
|
|
budget,request,result=prepare(tool_case)
|
|
evidence=EvidenceStore(store)
|
|
saved=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
changed=copy.deepcopy(result); changed["results"][0]["data"]["total_matches"]=999
|
|
with pytest.raises(EvidenceError,match="EVIDENCE_STEP_CONFLICT"):
|
|
evidence.store_tool_results(*CASE,"query1",request,changed)
|
|
assert evidence.snapshot(*CASE).package==saved.package
|
|
fresh=copy.deepcopy(request); fresh["requests"][0]["action_id"]="unreserved"
|
|
result=InvestigationToolRouter().query(saved.package,fresh)
|
|
with pytest.raises(EvidenceError,match="EVIDENCE_RESERVATION"):
|
|
evidence.store_tool_results(*CASE,"unreserved",fresh,result)
|
|
|
|
|
|
def test_deleted_index_cannot_reset_ids(tool_case):
|
|
budget,request,result=prepare(tool_case)
|
|
evidence=EvidenceStore(tool_case[0])
|
|
evidence.store_tool_results(*CASE,"query1",request,result)
|
|
tool_case[0].advance(*CASE,State.GENERATING_HYPOTHESES,step_id="deleted-index",artifacts={"evidence_index":None})
|
|
with pytest.raises(EvidenceError,match="EVIDENCE_POINTER_CHANGED"):
|
|
evidence.snapshot(*CASE)
|
|
|
|
|
|
def test_budget_rejects_wrong_action_evidence_binding(tool_case):
|
|
budget,request,result=prepare(tool_case)
|
|
saved=EvidenceStore(tool_case[0]).store_tool_results(*CASE,"query1",request,result)
|
|
with pytest.raises(BudgetError):
|
|
budget.record_evidence(*CASE,"H2","A1",saved.record_refs[0])
|
|
|
|
|
|
def test_limits_and_invalid_operation_ids_do_not_publish(tool_case):
|
|
budget,request,result=prepare(tool_case)
|
|
before=budget.snapshot(*CASE)
|
|
for limits in [EvidenceLimits(max_json_bytes=10),EvidenceLimits(max_total_bytes=10)]:
|
|
with pytest.raises(EvidenceError):
|
|
EvidenceStore(tool_case[0],limits).store_tool_results(*CASE,"query1",request,result)
|
|
for operation in [None,"",True]:
|
|
with pytest.raises(EvidenceError):
|
|
EvidenceStore(tool_case[0]).store_tool_results(*CASE,operation,request,result)
|
|
assert budget.snapshot(*CASE).package==before.package
|
|
|
|
|
|
def test_publication_index_crash_replay_does_not_create_new_ids(tool_case,monkeypatch):
|
|
budget,request,result=prepare(tool_case)
|
|
store=tool_case[0]; evidence=EvidenceStore(store)
|
|
original=store._index
|
|
def fail(path,manifest):
|
|
if manifest["artifacts"].get("evidence_index"):
|
|
raise OSError("Simulated crash after archive publication")
|
|
original(path,manifest)
|
|
monkeypatch.setattr(store,"_index",fail)
|
|
with pytest.raises(EvidenceError): evidence.store_tool_results(*CASE,"query1",request,result)
|
|
monkeypatch.setattr(store,"_index",original)
|
|
restored=evidence.store_tool_results(*CASE,"query1",request,result)
|
|
assert restored.reused and len(evidence.snapshot(*CASE).record_refs)==1
|
|
|
|
|
|
def test_confirmed_test_feedback_evidence_does_not_refill_budget(hypotheses):
|
|
store,budget,coordinator,before,request,feedback=async_setup(hypotheses)
|
|
coordinator.submit(request)
|
|
resumed=coordinator.confirm_feedback(feedback)
|
|
evidence=EvidenceStore(store)
|
|
saved=evidence.store_feedback(*CASE,"test-result",resumed.feedback_ref)
|
|
assert budget.snapshot(*CASE).state==before.state
|
|
budget.record_evidence(*CASE,"H1","test-action",saved.record_refs[0])
|
|
settled=budget.settle(*CASE,"H1","test-action",semantic_score=50,reason="Useful observation")
|
|
assert settled.state["accounts"]["H1"]["remaining"]==76
|
|
assert evidence.snapshot(*CASE).record_refs==saved.record_refs
|
|
|
|
|
|
@pytest.mark.parametrize("name",["evidence-content","evidence-record","evidence-index"])
|
|
def test_evidence_schemas_valid(name):
|
|
Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath(name+'-v1.schema.json').read_text()))
|