118 lines
6.4 KiB
Python
118 lines
6.4 KiB
Python
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.budget import BudgetConfig, BudgetManager
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, InboxProcessor, LifecycleError, State, Workspace
|
|
from viano_bug_analysis.lifecycle.async_tests import AsyncTestCoordinator
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_budget_manager import hypotheses
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case
|
|
from test_hypothesis_builder import interpreted_case
|
|
|
|
CASE = ("case-A", "stable-package")
|
|
|
|
|
|
def setup(hypotheses, executor="human", deadline="2099-01-01T00:00:00Z"):
|
|
store = hypotheses[0]
|
|
budget = BudgetManager(store, BudgetConfig(action_costs={"testing.request_human":40,"testing.request_automated":20}))
|
|
budget.initialize(*CASE)
|
|
action = "testing.request_human" if executor == "human" else "testing.request_automated"
|
|
budget.reserve(*CASE, "H1", "test-action", action)
|
|
store.advance(*CASE, State.INVESTIGATING_HYPOTHESES, step_id="begin-investigation")
|
|
before = budget.snapshot(*CASE)
|
|
request = {"format":"viano-async-test-request","format_version":1,"case_id":CASE[0],"package_id":CASE[1],
|
|
"analysis_revision":1,"request_id":"test-1","hypothesis_id":"H1","action_id":"test-action",
|
|
"executor":executor,"question":"What happens after interacting?","steps":["Interact once"],
|
|
"prerequisites":["Use the specified test environment"],"requested_observations":["Record the observed transition"],
|
|
"environment":"test-build-A","deadline":deadline,"budget_ledger_ref":before.event_ref}
|
|
feedback = {name:request[name] for name in ("format_version","case_id","package_id","analysis_revision","request_id","hypothesis_id","action_id","executor")}
|
|
feedback.update(format="viano-async-test-feedback",outcome="completed",observations=["The transition was absent"],
|
|
environment="test-build-A",tester_id="tester-1",attachments=[],
|
|
confirmation={"confirmed":True,"confirmed_by":"reviewer-1","confirmed_at":"2026-10-14T12:00:00Z"})
|
|
return store,budget,AsyncTestCoordinator(store),before,request,feedback
|
|
|
|
|
|
@pytest.mark.parametrize("executor", ["human","automated"])
|
|
def test_wait_confirmation_preserves_budget_revision_and_payloads(hypotheses, executor):
|
|
store,budget,coordinator,before,request,feedback = setup(hypotheses,executor)
|
|
blocked = coordinator.submit(request)
|
|
assert coordinator.submit(request).reused
|
|
assert budget.snapshot(*CASE).state == before.state
|
|
with PackageReader(blocked.package) as reader:
|
|
assert reader.manifest["state"] == State.BLOCKED
|
|
assert reader.manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK"
|
|
with pytest.raises(LifecycleError, match="Confirmed test feedback"):
|
|
store.advance(*CASE,State.INVESTIGATING_HYPOTHESES,step_id="timer-resume")
|
|
resumed = coordinator.confirm_feedback(feedback)
|
|
assert coordinator.confirm_feedback(feedback).reused
|
|
assert budget.snapshot(*CASE).state == before.state
|
|
assert PackageValidator().validate(resumed.package).valid
|
|
with PackageReader(before.package) as old, PackageReader(resumed.package) as new:
|
|
assert new.manifest["state"] == State.INVESTIGATING_HYPOTHESES
|
|
assert new.manifest["analysis_revision"] == 1
|
|
assert new.manifest["package_generation"] == old.manifest["package_generation"] + 2
|
|
assert new.manifest["routing"]["retry_not_before"] is None
|
|
for name in old.files:
|
|
if name != "manifest.json":
|
|
assert old.read_bytes(name) == new.read_bytes(name)
|
|
# Feedback is an investigation result, not a reason to regenerate the case.
|
|
assert hypotheses[1].build(*CASE).reused
|
|
budget.record_result(*CASE,"H1","test-action",{"observations":feedback["observations"]},source_refs=[resumed.feedback_ref])
|
|
assessed = budget.settle(*CASE,"H1","test-action",semantic_score=50,reason="Useful observation")
|
|
cost = 40 if executor == "human" else 20
|
|
assert assessed.state["accounts"]["H1"]["remaining"] == 100-cost+cost*4//10
|
|
|
|
|
|
def test_deadline_does_not_resume_or_charge_idle_case(hypotheses):
|
|
store,budget,coordinator,before,request,feedback = setup(hypotheses,deadline="2000-01-01T00:00:00Z")
|
|
coordinator.submit(request)
|
|
outcomes = InboxProcessor(store.workspace).poll()
|
|
assert not any(outcome.status == "RESUMED" for outcome in outcomes)
|
|
assert budget.snapshot(*CASE).state == before.state
|
|
coordinator.confirm_feedback(feedback)
|
|
assert budget.snapshot(*CASE).state == before.state
|
|
|
|
|
|
@pytest.mark.parametrize("change", [
|
|
lambda f: f.update(analysis_revision=2),
|
|
lambda f: f.update(hypothesis_id="H2"),
|
|
lambda f: f.update(request_id="other"),
|
|
lambda f: f["confirmation"].update(confirmed=False),
|
|
lambda f: f.update(attachments=[{"source_ref":"../escape","sha256":"0"*64}]),
|
|
])
|
|
def test_invalid_feedback_never_wakes_or_changes_budget(hypotheses, change):
|
|
store,budget,coordinator,before,request,feedback = setup(hypotheses)
|
|
blocked = coordinator.submit(request)
|
|
change(feedback)
|
|
with pytest.raises(LifecycleError):
|
|
coordinator.confirm_feedback(feedback)
|
|
assert budget.snapshot(*CASE).package == blocked.package
|
|
assert budget.snapshot(*CASE).state == before.state
|
|
|
|
|
|
def test_portable_wait_feedback_and_conflicting_replay(hypotheses,tmp_path):
|
|
store,budget,coordinator,before,request,feedback = setup(hypotheses)
|
|
blocked = coordinator.submit(request)
|
|
portable = CheckpointStore(Workspace(tmp_path / "other-service"))
|
|
status, managed = portable.intake(blocked.package)
|
|
assert status == "ACCEPTED"
|
|
assert BudgetManager(portable).snapshot(*CASE).state == before.state
|
|
remote_resumed = AsyncTestCoordinator(portable).confirm_feedback(feedback)
|
|
with PackageReader(remote_resumed.package) as reader:
|
|
assert reader.manifest["state"] == State.INVESTIGATING_HYPOTHESES
|
|
with PackageReader(blocked.package) as reader:
|
|
assert reader.manifest["artifacts"]["async_test_request"] == blocked.request_ref
|
|
changed = copy.deepcopy(request)
|
|
changed["question"] = "A different test"
|
|
with pytest.raises(LifecycleError):
|
|
coordinator.submit(changed)
|
|
resumed = coordinator.confirm_feedback(feedback)
|
|
changed = copy.deepcopy(feedback)
|
|
changed["observations"] = ["Different feedback"]
|
|
with pytest.raises(LifecycleError):
|
|
coordinator.confirm_feedback(changed)
|
|
assert budget.snapshot(*CASE).package == resumed.package
|