436 lines
22 KiB
Python
436 lines
22 KiB
Python
import asyncio
|
|
import hashlib
|
|
import json
|
|
from importlib.resources import files
|
|
from zipfile import ZipFile
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.analysis import HypothesisBuilder, HypothesisConfig
|
|
from viano_bug_analysis.budget import BudgetConfig, BudgetError, BudgetLimits, BudgetManager
|
|
from viano_bug_analysis.budget.engine import apply, initial_state
|
|
from viano_bug_analysis.budget.scoring import reward_for
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, State, Workspace
|
|
from viano_bug_analysis.lifecycle.errors import LockBusy
|
|
from viano_bug_analysis.lifecycle.locking import FileLock
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes, key
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case, gateway
|
|
from test_hypothesis_builder import interpreted_case, ProposalAdapter
|
|
|
|
|
|
CASE = ("case-A", "stable-package")
|
|
|
|
|
|
@pytest.fixture
|
|
def hypotheses(interpreted_case):
|
|
store, case_builder, _, _, _ = interpreted_case
|
|
adapter = ProposalAdapter()
|
|
result = asyncio.run(HypothesisBuilder(store, gateway(adapter)).build(*CASE))
|
|
return store, case_builder, result, adapter
|
|
|
|
|
|
def manager(hypotheses, config=None):
|
|
store, _, result, _ = hypotheses
|
|
budget = BudgetManager(store, config)
|
|
budget.initialize(*CASE)
|
|
with PackageReader(result.package) as reader:
|
|
source = json.loads(reader.read_bytes(reader.manifest["artifacts"]["case"]))["inputs"]["records"][0]["path"]
|
|
return budget, source
|
|
|
|
|
|
def result_and_settle(budget, source, action_id, evidence=None, score=100, hypothesis="H1"):
|
|
budget.record_result(*CASE, hypothesis, action_id, {"finding": action_id} if evidence is None else evidence, source_refs=[source])
|
|
return budget.settle(*CASE, hypothesis, action_id, semantic_score=score, reason="Controller assessment of stored result")
|
|
|
|
|
|
def test_durable_debit_reward_preservation_and_portability(hypotheses, tmp_path):
|
|
budget, source = manager(hypotheses)
|
|
initial = budget.snapshot(*CASE)
|
|
assert set(initial.state["accounts"]) == {"H1", "H2", "H3"}
|
|
assert {a["remaining"] for a in initial.state["accounts"].values()} == {100}
|
|
assert budget.reserve(*CASE, "H1", "query1", "source.find_text").state["accounts"]["H1"]["remaining"] == 90
|
|
settled = result_and_settle(budget, source, "query1")
|
|
account = settled.state["accounts"]["H1"]
|
|
assert (account["remaining"], account["spent"], account["rewarded"]) == (98, 10, 8)
|
|
assert account["actions"]["query1"]["status"] == "settled"
|
|
assert account["actions"]["query1"]["reward"] < account["actions"]["query1"]["cost"]
|
|
assert settled.investigable == ["H1", "H2", "H3"]
|
|
assert PackageValidator().validate(settled.package).valid
|
|
with PackageReader(hypotheses[2].package) as before, PackageReader(settled.package) as after:
|
|
assert after.manifest["package_generation"] == before.manifest["package_generation"] + 4
|
|
assert after.manifest["analysis_revision"] == 1
|
|
for name in before.files:
|
|
if name != "manifest.json":
|
|
assert before.read_bytes(name) == after.read_bytes(name)
|
|
event = json.loads(after.read_bytes(settled.event_ref))
|
|
assert event["request"]["semantic_score"] == 100
|
|
assert event["state"] == settled.state
|
|
assert after.manifest["artifacts"]["diagnosis"] is None
|
|
portable = CheckpointStore(Workspace(tmp_path / "fresh-service"))
|
|
renamed = tmp_path / "renamed.zip"
|
|
renamed.write_bytes(settled.package.read_bytes())
|
|
portable.intake(renamed)
|
|
assert BudgetManager(portable).snapshot(*CASE).state == settled.state
|
|
assert hypotheses[1].build(*CASE).reused
|
|
assert asyncio.run(HypothesisBuilder(hypotheses[0], gateway(hypotheses[3])).build(*CASE)).reused
|
|
|
|
|
|
def test_idempotent_operations_and_conflict(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
reserved = budget.reserve(*CASE, "H1", "q", "read")
|
|
assert budget.reserve(*CASE, "H1", "q", "read").reused
|
|
assert budget.snapshot(*CASE).package == reserved.package
|
|
with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"):
|
|
budget.reserve(*CASE, "H1", "q", "different")
|
|
first = budget.record_result(*CASE, "H1", "q", {"b": 2, "a": 1}, source_refs=[source])
|
|
assert budget.record_result(*CASE, "H1", "q", {"a": 1, "b": 2}, source_refs=[source]).package == first.package
|
|
with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"):
|
|
budget.record_result(*CASE, "H1", "q", {"a": 3}, source_refs=[source])
|
|
settled = budget.settle(*CASE, "H1", "q", semantic_score=50, reason="Useful")
|
|
assert budget.settle(*CASE, "H1", "q", semantic_score=50, reason="Useful").package == settled.package
|
|
with pytest.raises(BudgetError, match="BUDGET_STEP_CONFLICT"):
|
|
budget.settle(*CASE, "H1", "q", semantic_score=100, reason="Useful")
|
|
assert budget.initialize(*CASE).state["accounts"]["H1"]["remaining"] == 94
|
|
# Python equality must not let a float masquerade as a valid replayed score.
|
|
with pytest.raises(BudgetError):
|
|
budget.settle(*CASE, "H1", "q", semantic_score=50.0, reason="Useful")
|
|
|
|
|
|
@pytest.mark.parametrize("evidence", [{1:"invalid JSON object key"}, {"value":float("nan")}, "unstructured"])
|
|
def test_invalid_evidence_does_not_publish_partial_result(hypotheses, evidence):
|
|
budget, source = manager(hypotheses)
|
|
before = budget.reserve(*CASE, "H1", "q", "read")
|
|
with pytest.raises(BudgetError):
|
|
budget.record_result(*CASE, "H1", "q", evidence, source_refs=[source])
|
|
assert budget.snapshot(*CASE).package == before.package
|
|
|
|
|
|
def test_duplicate_evidence_no_reward_even_across_hypotheses(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "first", "read")
|
|
result_and_settle(budget, source, "first", {"finding": "same"})
|
|
budget.reserve(*CASE, "H2", "other", "different_query")
|
|
settled = result_and_settle(budget, source, "other", {"finding": "same"}, hypothesis="H2")
|
|
action = settled.state["accounts"]["H2"]["actions"]["other"]
|
|
assert action["novelty_score"] == action["reward"] == 0
|
|
assert len(settled.state["seen_evidence"]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("score", [-1, 101, True, 50.5, 100.0, "100", None])
|
|
def test_invalid_semantic_scores_atomic(hypotheses, score):
|
|
budget, source = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
before = budget.record_result(*CASE, "H1", "q", {"finding": "new"}, source_refs=[source])
|
|
with pytest.raises(BudgetError):
|
|
budget.settle(*CASE, "H1", "q", semantic_score=score, reason="Value")
|
|
assert budget.snapshot(*CASE).package == before.package
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 90
|
|
|
|
|
|
def test_low_information_stops_and_cannot_be_refilled(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
for number in range(3):
|
|
budget.reserve(*CASE, "H1", f"q{number}", "read")
|
|
result_and_settle(budget, source, f"q{number}", score=20)
|
|
state = budget.snapshot(*CASE).state
|
|
assert state["accounts"]["H1"]["status"] == "abandoned_low_information_gain"
|
|
assert state["accounts"]["H1"]["low_information_steps"] == 3
|
|
with pytest.raises(BudgetError, match="BUDGET_TERMINAL"):
|
|
budget.reserve(*CASE, "H1", "q4", "read")
|
|
assert BudgetManager(hypotheses[0], BudgetConfig(max_low_information_steps=99)).initialize(*CASE).state == state
|
|
|
|
|
|
def test_high_information_resets_consecutive_counter(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
for number, score in enumerate([0, 0, 100, 0, 0]):
|
|
budget.reserve(*CASE, "H1", str(number), "read")
|
|
result_and_settle(budget, source, str(number), score=score)
|
|
account = budget.snapshot(*CASE).state["accounts"]["H1"]
|
|
assert account["status"] == "investigating" and account["low_information_steps"] == 2
|
|
|
|
|
|
def test_failed_action_consumes_cost_and_pending_survives_restart(hypotheses):
|
|
budget, _ = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
restarted = BudgetManager(hypotheses[0])
|
|
assert "H1" not in restarted.snapshot(*CASE).investigable
|
|
with pytest.raises(BudgetError, match="BUDGET_PENDING_ACTION"):
|
|
restarted.reserve(*CASE, "H1", "q2", "read")
|
|
with pytest.raises(BudgetError, match="BUDGET_PENDING_ACTION"):
|
|
restarted.reject(*CASE, "H1", reason="Rejected")
|
|
failed = restarted.fail(*CASE, "H1", "q", reason="Timeout; outcome uncertain")
|
|
account = failed.state["accounts"]["H1"]
|
|
assert account["remaining"] == 90 and account["rewarded"] == 0
|
|
assert restarted.fail(*CASE, "H1", "q", reason="Timeout; outcome uncertain").reused
|
|
|
|
|
|
def test_cost_overrides_and_exhaustion(hypotheses):
|
|
config = BudgetConfig(default_action_cost=40, action_costs={"expensive": 101}, max_reward_bps=0, max_low_information_steps=50)
|
|
budget, source = manager(hypotheses, config)
|
|
before = budget.snapshot(*CASE)
|
|
with pytest.raises(BudgetError, match="BUDGET_INSUFFICIENT"):
|
|
budget.reserve(*CASE, "H1", "too_much", "expensive")
|
|
assert budget.snapshot(*CASE).package == before.package
|
|
for action_id in ["a", "b"]:
|
|
budget.reserve(*CASE, "H1", action_id, "read")
|
|
result_and_settle(budget, source, action_id)
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["status"] == "budget_exhausted"
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 20
|
|
|
|
|
|
def test_reward_can_restore_affordability_after_reservation(hypotheses):
|
|
budget, source = manager(hypotheses, BudgetConfig(default_action_cost=60, max_reward_bps=8000))
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["status"] == "investigating"
|
|
account = result_and_settle(budget, source, "q").state["accounts"]["H1"]
|
|
assert account["remaining"] == 88 and account["status"] == "investigating"
|
|
|
|
|
|
@pytest.mark.parametrize("evidence", [{}, []])
|
|
def test_empty_results_not_rewarded(hypotheses, evidence):
|
|
budget, source = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
assert result_and_settle(budget, source, "q", evidence).state["accounts"]["H1"]["rewarded"] == 0
|
|
|
|
|
|
def test_terminal_state_is_separate_from_confidence_and_creation_registry(hypotheses):
|
|
budget, _ = manager(hypotheses)
|
|
terminal = budget.reject(*CASE, "H1", reason="Evidence refutes mechanism")
|
|
assert terminal.state["accounts"]["H1"]["status"] == "rejected"
|
|
assert "H1" not in terminal.investigable
|
|
with pytest.raises(BudgetError, match="BUDGET_TERMINAL"):
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
with PackageReader(hypotheses[2].package) as before, PackageReader(terminal.package) as after:
|
|
ref = after.manifest["artifacts"]["hypothesis_registry"]
|
|
assert after.read_bytes(ref) == before.read_bytes(ref)
|
|
assert json.loads(after.read_bytes(after.manifest["artifacts"]["hypotheses"]))["hypotheses"][0]["confidence"] == "unassessed"
|
|
|
|
|
|
def test_policy_is_frozen_and_mutation_rejected(hypotheses):
|
|
budget, _ = manager(hypotheses)
|
|
changed = BudgetManager(hypotheses[0], BudgetConfig(max_reward_bps=9999))
|
|
assert changed.initialize(*CASE).reused
|
|
with pytest.raises(BudgetError, match="BUDGET_POLICY_CHANGED"):
|
|
changed.reserve(*CASE, "H1", "q", "read")
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 100
|
|
|
|
|
|
@pytest.mark.parametrize("config", [{"default_action_cost":0}, {"max_reward_bps":10000}, {"max_reward_bps":True},
|
|
{"new_evidence_score":101}, {"low_information_threshold":-1}, {"max_low_information_steps":0},
|
|
{"action_costs":{"x":False}}, {"action_costs":{" ":1}}, {"default_action_cost":1.5}])
|
|
def test_configuration_rejected(config):
|
|
with pytest.raises(BudgetError):
|
|
BudgetConfig(**config)
|
|
|
|
|
|
def test_config_toml_and_dictionary_detachment(tmp_path):
|
|
config = BudgetConfig.load("config/budget.example.toml")
|
|
assert config.action_costs["source.read_context"] == 5
|
|
values = {"read":5}
|
|
config = BudgetConfig(action_costs=values)
|
|
values["read"] = 0
|
|
assert config.policy()["action_costs"]["read"] == 5
|
|
invalid = tmp_path / "bad.toml"
|
|
invalid.write_text('[budget]\nunknown=1')
|
|
with pytest.raises(BudgetError):
|
|
BudgetConfig.load(invalid)
|
|
with pytest.raises(BudgetError):
|
|
BudgetLimits(max_events=True)
|
|
|
|
|
|
def test_reward_strictly_below_cost_and_both_scores_required():
|
|
for cost in range(1, 101):
|
|
for novelty in [0, 20, 100]:
|
|
for semantic in [0, 20, 100]:
|
|
reward, information = reward_for(cost, novelty, semantic, BudgetConfig(max_reward_bps=9999).policy())
|
|
assert 0 <= reward < cost
|
|
assert information == min(novelty, semantic)
|
|
if min(novelty, semantic) == 0:
|
|
assert reward == 0
|
|
assert reward_for(10, 50, 100, BudgetConfig().policy()) == (4, 50)
|
|
|
|
|
|
def test_finite_steps_even_with_maximum_rewards():
|
|
policy = BudgetConfig(default_action_cost=1, max_reward_bps=9999).policy()
|
|
state = initial_state({"creations":[{"id":"H1","allocated_budget":10}]}, policy)
|
|
for index in range(10):
|
|
action_id = str(index)
|
|
state = apply(state, {"operation":"reserve","hypothesis_id":"H1","action_id":action_id,"action":"read"}, policy)
|
|
state = apply(state, {"operation":"result","hypothesis_id":"H1","action_id":action_id,"evidence":{"n":index},"result_ref":"test"}, policy)
|
|
state = apply(state, {"operation":"settle","hypothesis_id":"H1","action_id":action_id,"semantic_score":100,"reason":"New"}, policy)
|
|
assert state["accounts"]["H1"]["remaining"] == 0
|
|
assert state["accounts"]["H1"]["status"] == "budget_exhausted"
|
|
|
|
|
|
def test_lifecycle_preconditions(interpreted_case):
|
|
with pytest.raises(BudgetError, match="BUDGET_HYPOTHESES_NOT_READY"):
|
|
BudgetManager(interpreted_case[0]).initialize(*CASE)
|
|
|
|
|
|
def test_missing_init_result_and_bad_references_rejected(hypotheses):
|
|
budget = BudgetManager(hypotheses[0])
|
|
with pytest.raises(BudgetError, match="BUDGET_NOT_INITIALIZED"):
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
budget.initialize(*CASE)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
with pytest.raises(BudgetError, match="BUDGET_RESULT_REQUIRED"):
|
|
budget.settle(*CASE, "H1", "q", semantic_score=100, reason="Missing")
|
|
for refs in [[], ["../escape"], ["missing.json"], ["manifest.json"]]:
|
|
with pytest.raises(BudgetError):
|
|
budget.record_result(*CASE, "H1", "q", {"x":1}, source_refs=refs)
|
|
with pytest.raises(BudgetError, match="BUDGET_HYPOTHESIS_UNKNOWN"):
|
|
budget.reserve(*CASE, "H99", "q", "read")
|
|
|
|
|
|
def test_limits_do_not_truncate_or_drop_accounting(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
small = BudgetManager(hypotheses[0], limits=BudgetLimits(max_events=1))
|
|
with pytest.raises(BudgetError, match="BUDGET_HISTORY_LIMIT"):
|
|
small.reserve(*CASE, "H1", "q", "read")
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
small = BudgetManager(hypotheses[0], limits=BudgetLimits(max_result_bytes=5))
|
|
with pytest.raises(BudgetError, match="BUDGET_RESULT_LIMIT"):
|
|
small.record_result(*CASE, "H1", "q", {"long":"content"}, source_refs=[source])
|
|
|
|
|
|
def test_publish_index_crash_recovered_without_double_debit(hypotheses, monkeypatch):
|
|
budget, _ = manager(hypotheses)
|
|
original = budget.store._index
|
|
def fail_index(*args):
|
|
raise OSError("Simulated index failure after publication")
|
|
monkeypatch.setattr(budget.store, "_index", fail_index)
|
|
with pytest.raises(BudgetError):
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
monkeypatch.setattr(budget.store, "_index", original)
|
|
recovered = budget.reserve(*CASE, "H1", "q", "read")
|
|
assert recovered.reused and recovered.state["accounts"]["H1"]["remaining"] == 90
|
|
index = json.loads((recovered.package.parent / "current.json").read_bytes())
|
|
assert index["package"] == recovered.package.name
|
|
|
|
|
|
def test_same_case_lock_and_concurrent_generation_guard(hypotheses, monkeypatch):
|
|
budget, _ = manager(hypotheses)
|
|
with FileLock(budget.store.workspace.runtime / "locks" / f"budget-{key(CASE[0])}.lock"):
|
|
with pytest.raises(LockBusy):
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
original = budget.store.advance
|
|
def race(*args, **kwargs):
|
|
original(*CASE, State.GENERATING_HYPOTHESES, step_id="concurrent-note")
|
|
return original(*args, **kwargs)
|
|
monkeypatch.setattr(budget.store, "advance", race)
|
|
with pytest.raises(BudgetError, match="GENERATION_CHANGED"):
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
assert budget.snapshot(*CASE).state["accounts"]["H1"]["remaining"] == 100
|
|
|
|
|
|
def test_pointer_deletion_does_not_reset_accounting(hypotheses):
|
|
budget, _ = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
budget.store.advance(*CASE, State.GENERATING_HYPOTHESES, step_id="bad-pointer", artifacts={"budget_ledger":None})
|
|
with pytest.raises(BudgetError, match="BUDGET_POINTER_CHANGED"):
|
|
budget.initialize(*CASE)
|
|
|
|
|
|
def rewrite_managed_fixture(path, mutate):
|
|
"""Simulate an imported corrupt ledger whose archive integrity was recomputed."""
|
|
with ZipFile(path) as archive:
|
|
payloads = {name: archive.read(name) for name in archive.namelist()}
|
|
manifest = json.loads(payloads["manifest.json"])
|
|
event_ref = manifest["artifacts"]["budget_ledger"]
|
|
cp_ref = manifest["artifacts"]["lifecycle_checkpoint"]
|
|
event, checkpoint = json.loads(payloads[event_ref]), json.loads(payloads[cp_ref])
|
|
mutate(event, checkpoint)
|
|
payloads[event_ref] = json_bytes(event)
|
|
checkpoint["details"]["event_sha256"] = hashlib.sha256(payloads[event_ref]).hexdigest()
|
|
payloads[cp_ref] = json_bytes(checkpoint)
|
|
for ref in (event_ref, cp_ref):
|
|
manifest["integrity"][ref] = {"sha256": hashlib.sha256(payloads[ref]).hexdigest(), "size": len(payloads[ref])}
|
|
payloads["manifest.json"] = json_bytes(manifest)
|
|
temporary = path.with_suffix(".tampered.zip")
|
|
with ZipFile(temporary, "w") as archive:
|
|
for name, content in payloads.items():
|
|
archive.writestr(name, content)
|
|
temporary.replace(path)
|
|
assert PackageValidator().validate(path).valid
|
|
|
|
|
|
@pytest.mark.parametrize("mutate", [
|
|
lambda e,c: e["state"]["accounts"]["H1"].update(remaining=100),
|
|
lambda e,c: e["state"]["accounts"]["H1"]["actions"]["q"].update(cost=1),
|
|
lambda e,c: e.update(parent_sha256="0"*64),
|
|
lambda e,c: e["policy"].update(max_reward_bps=9999),
|
|
lambda e,c: e.update(registry_sha256="0"*64),
|
|
])
|
|
def test_recomputed_archive_hashes_cannot_hide_ledger_corruption(hypotheses, mutate):
|
|
budget, _ = manager(hypotheses)
|
|
reserved = budget.reserve(*CASE, "H1", "q", "read")
|
|
rewrite_managed_fixture(reserved.package, mutate)
|
|
with pytest.raises(BudgetError):
|
|
budget.snapshot(*CASE)
|
|
|
|
|
|
def test_reward_tampering_rejected_by_replay(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
settled = result_and_settle(budget, source, "q")
|
|
def mutate(event, checkpoint):
|
|
account = event["state"]["accounts"]["H1"]
|
|
account.update(remaining=100, rewarded=10)
|
|
account["actions"]["q"]["reward"] = 10
|
|
rewrite_managed_fixture(settled.package, mutate)
|
|
with pytest.raises(BudgetError, match="BUDGET_ACCOUNTING_CHANGED"):
|
|
budget.snapshot(*CASE)
|
|
|
|
|
|
def test_zero_affordable_actions_ready_for_evaluation(hypotheses):
|
|
budget = BudgetManager(hypotheses[0], BudgetConfig(default_action_cost=101))
|
|
snapshot = budget.initialize(*CASE)
|
|
assert snapshot.ready_for_evaluation
|
|
assert not snapshot.investigable
|
|
assert {a["status"] for a in snapshot.state["accounts"].values()} == {"budget_exhausted"}
|
|
|
|
|
|
def test_pending_actions_not_ready_for_evaluation(hypotheses):
|
|
budget, _ = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
budget.reject(*CASE, "H2", reason="Refuted")
|
|
budget.reject(*CASE, "H3", reason="Refuted")
|
|
assert not budget.snapshot(*CASE).investigable
|
|
assert not budget.snapshot(*CASE).ready_for_evaluation
|
|
|
|
|
|
def test_new_revision_new_policy_without_losing_old_bytes(hypotheses):
|
|
budget, source = manager(hypotheses)
|
|
budget.reserve(*CASE, "H1", "q", "read")
|
|
result_and_settle(budget, source, "q", {"finding":"same"})
|
|
before = budget.snapshot(*CASE)
|
|
store, case_builder, _, _ = hypotheses
|
|
store.advance(*CASE, State.FAILED_TO_ANALYZE, step_id="review", reason="Needs evidence")
|
|
store.advance(*CASE, State.REOPENED, step_id="new-revision")
|
|
case_builder.build(*CASE)
|
|
from test_case_interpreter import FixtureAdapter
|
|
from viano_bug_analysis.analysis import CaseInterpreter
|
|
asyncio.run(CaseInterpreter(store, gateway(FixtureAdapter())).interpret(*CASE))
|
|
asyncio.run(HypothesisBuilder(store, gateway(ProposalAdapter()), HypothesisConfig(initial_budget=23)).build(*CASE))
|
|
new_budget = BudgetManager(store, BudgetConfig(default_action_cost=3))
|
|
after = new_budget.initialize(*CASE)
|
|
assert after.analysis_revision == 2
|
|
assert {a["remaining"] for a in after.state["accounts"].values()} == {23}
|
|
with PackageReader(before.package) as old, PackageReader(after.package) as new:
|
|
for name in old.files:
|
|
if name != "manifest.json":
|
|
assert old.read_bytes(name) == new.read_bytes(name)
|
|
new_budget.reserve(*CASE, "H1", "q", "read")
|
|
duplicate = result_and_settle(new_budget, source, "q", {"finding":"same"})
|
|
assert duplicate.state["accounts"]["H1"]["rewarded"] == 0
|
|
assert BudgetManager(store, BudgetConfig(default_action_cost=3)).snapshot(*CASE).state == duplicate.state
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["budget-policy", "budget-request", "budget-event"])
|
|
def test_schemas_well_formed(name):
|
|
schema = json.loads(files("viano_bug_analysis.schemas").joinpath(name + "-v1.schema.json").read_text())
|
|
Draft202012Validator.check_schema(schema)
|