309 lines
16 KiB
Python
309 lines
16 KiB
Python
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, LifecycleError, State, Workspace
|
|
from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER, checkpoint_chain
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter
|
|
|
|
|
|
@pytest.fixture
|
|
def admitted(make_package, tmp_path):
|
|
workspace = Workspace(tmp_path / "service")
|
|
store = CheckpointStore(workspace)
|
|
status, path = store.intake(make_package(entries={"unknown": b"keep"}))
|
|
assert status == "ACCEPTED"
|
|
return workspace, store, path
|
|
|
|
|
|
def test_checkpoint_inside_package_and_runtime_is_coarse(admitted):
|
|
workspace, store, path = admitted
|
|
with PackageReader(path) as reader:
|
|
manifest = reader.manifest
|
|
assert manifest["state"] == State.VALIDATED
|
|
assert manifest["package_generation"] == 1
|
|
assert manifest["analysis_revision"] == 0
|
|
chain = checkpoint_chain(reader)
|
|
assert chain[0][1]["previous_state"] == State.NEW
|
|
assert chain[0][1]["source_generation"] == 0
|
|
assert chain[0][1]["details"] == {"operation": "intake"}
|
|
assert reader.read_bytes("unknown") == b"keep"
|
|
assert json.loads((workspace.runtime / "current_job.json").read_text()) is None
|
|
|
|
|
|
def test_progress_idempotence_and_conflicting_step(admitted):
|
|
_, store, _ = admitted
|
|
path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["package_generation"] == 2
|
|
assert reader.manifest["analysis_revision"] == 1
|
|
assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == path
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"different": True})
|
|
assert error.value.code == "STEP_CONFLICT"
|
|
next_path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-progress", details={"collected": ["input-A"]}, additions={"case/inputs/a": b"generic"})
|
|
assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == next_path
|
|
with PackageReader(next_path) as reader:
|
|
assert len(checkpoint_chain(reader)) == 3
|
|
assert reader.manifest["analysis_revision"] == 1
|
|
assert reader.read_bytes("case/inputs/a") == b"generic"
|
|
|
|
|
|
def test_full_lifecycle_and_reopen(admitted):
|
|
_, store, _ = admitted
|
|
states = [State.BUILDING_CASE, State.INTERPRETING_CASE, State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES, State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF, State.REOPENED, State.BUILDING_CASE]
|
|
for index, state in enumerate(states):
|
|
path = store.advance("case-A", "stable-package", state, step_id=f"step-{index}")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["analysis_revision"] == 2
|
|
assert reader.manifest["history"]["analysis_revisions"] == 2
|
|
assert reader.manifest["package_generation"] == 10
|
|
assert len(checkpoint_chain(reader)) == 10
|
|
assert PackageValidator().validate(path).valid
|
|
|
|
|
|
def test_failure_routes_review_then_reopen(admitted):
|
|
_, store, _ = admitted
|
|
path = store.advance("case-A", "stable-package", State.FAILED_TO_ANALYZE, step_id="failure", reason="No available analysis")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["routing"]["destination"] == "review"
|
|
path = store.advance("case-A", "stable-package", State.REOPENED, step_id="review-return")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["analysis_revision"] == 1
|
|
|
|
|
|
def test_block_retry_and_resume(admitted):
|
|
_, store, _ = admitted
|
|
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
|
path = store.advance("case-A", "stable-package", State.BLOCKED, step_id="blocked", reason="External dependency", retry_not_before=future)
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["routing"]["retry_count"] == 1
|
|
assert checkpoint_chain(reader)[0][1]["resume_state"] == State.VALIDATED
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.VALIDATED, step_id="resume")
|
|
assert error.value.code == "RETRY_NOT_DUE"
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="wrong-resume")
|
|
|
|
|
|
@pytest.mark.parametrize("retry", [None, "not-a-date", "2026-01-01T00:00:00", "2026-02-30T00:00:00Z"])
|
|
def test_invalid_retry_rejected_before_publication(retry, admitted):
|
|
_, store, path = admitted
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason="External dependency", retry_not_before=retry)
|
|
assert store.latest("case-A", "stable-package") == path
|
|
|
|
|
|
def test_restart_uses_package_not_current_index(admitted):
|
|
workspace, _, path = admitted
|
|
(path.parent / "current.json").write_text("broken index")
|
|
(workspace.runtime / "current_job.json").write_text("broken runtime summary")
|
|
restarted = CheckpointStore(workspace)
|
|
assert restarted.latest("case-A", "stable-package") == path
|
|
assert restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start").exists()
|
|
|
|
|
|
def test_crash_after_package_publication_before_index(admitted, monkeypatch):
|
|
workspace, store, original = admitted
|
|
def crash(*args):
|
|
raise LifecycleError("SIMULATED_CRASH", "after publication")
|
|
monkeypatch.setattr(store, "_index", crash)
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
summary = json.loads((workspace.runtime / "current_job.json").read_text())
|
|
assert set(summary) == {"case_id", "package_id", "package_generation", "analysis_revision", "stage", "started_at", "last_checkpoint_at"}
|
|
restarted = CheckpointStore(workspace)
|
|
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
assert path != original
|
|
assert len(list(path.parent.glob("g*.zip"))) == 2
|
|
assert json.loads((workspace.runtime / "current_job.json").read_text()) is None
|
|
|
|
|
|
def test_crash_before_publication_repeats_only_pending_step(admitted, monkeypatch):
|
|
workspace, store, original = admitted
|
|
def crash(*args, **kwargs):
|
|
raise LifecycleError("SIMULATED_CRASH", "before publication")
|
|
monkeypatch.setattr(store.writer, "write", crash)
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
restarted = CheckpointStore(workspace)
|
|
assert restarted.latest("case-A", "stable-package") == original
|
|
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["package_generation"] == 2
|
|
|
|
|
|
def test_import_portable_package_without_service_memory(admitted, tmp_path):
|
|
_, store, _ = admitted
|
|
original = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"})
|
|
other = CheckpointStore(Workspace(tmp_path / "other-service"))
|
|
status, imported = other.intake(original)
|
|
assert status == "ACCEPTED"
|
|
with PackageReader(imported) as reader:
|
|
assert checkpoint_chain(reader)[1][1]["details"] == {"progress": "retained"}
|
|
# Previous work remains idempotent even without the previous service's files.
|
|
assert other.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"}) == imported
|
|
|
|
|
|
def test_newer_return_preserves_local_history(admitted, tmp_path):
|
|
_, store, original = admitted
|
|
returned = tmp_path / "returned"
|
|
PackageWriter().write(original, returned, additions={"feedback/new": b"new feedback"})
|
|
status, path = store.intake(returned)
|
|
assert status == "ACCEPTED"
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["package_generation"] == 3
|
|
assert reader.read_bytes("feedback/new") == b"new feedback"
|
|
|
|
|
|
def test_invalid_details_not_published(admitted):
|
|
_, store, original = admitted
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="invalid", details=["wrong"])
|
|
assert store.latest("case-A", "stable-package") == original
|
|
|
|
|
|
def test_corrupt_managed_package_fails_without_rollback(admitted):
|
|
_, store, path = admitted
|
|
newer = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start")
|
|
newer.write_bytes(b"broken")
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.latest("case-A", "stable-package")
|
|
assert error.value.code == "STORAGE_PACKAGE_INVALID"
|
|
assert path.exists()
|
|
|
|
|
|
def test_head_is_self_contained_while_full_history_audit_is_explicit(admitted):
|
|
workspace,store,older=admitted
|
|
head=store.advance('case-A','stable-package',State.BUILDING_CASE,step_id='new-head')
|
|
older.write_bytes(b'broken historical snapshot')
|
|
restarted=CheckpointStore(workspace)
|
|
assert restarted.latest('case-A','stable-package')==head
|
|
with PackageReader(head) as reader:
|
|
assert len(checkpoint_chain(reader))==2 and reader.read_bytes('unknown')==b'keep'
|
|
with pytest.raises(LifecycleError) as error:
|
|
restarted.latest('case-A','stable-package',audit_history=True)
|
|
assert error.value.code=='STORAGE_PACKAGE_INVALID'
|
|
|
|
|
|
def test_normal_head_lookup_does_not_read_obsolete_archive_bytes(admitted,monkeypatch):
|
|
from pathlib import Path
|
|
workspace,store,older=admitted
|
|
head=store.advance('case-A','stable-package',State.BUILDING_CASE,step_id='head-only')
|
|
reads=[];original=Path.open
|
|
def tracked(path,*args,**kwargs):
|
|
if path.suffix=='.zip':reads.append(path)
|
|
return original(path,*args,**kwargs)
|
|
monkeypatch.setattr(Path,'open',tracked)
|
|
restarted=CheckpointStore(workspace)
|
|
assert restarted.latest('case-A','stable-package')==head
|
|
assert reads and all(path==head for path in reads)
|
|
|
|
|
|
def test_cached_head_is_rechecked_even_with_identical_size_and_mtime(admitted):
|
|
import os
|
|
_,store,head=admitted
|
|
assert store.latest('case-A','stable-package')==head
|
|
info=head.stat();head.write_bytes(b'x'*info.st_size)
|
|
os.utime(head,ns=(info.st_atime_ns,info.st_mtime_ns))
|
|
with pytest.raises(LifecycleError) as error:store.latest('case-A','stable-package')
|
|
assert error.value.code=='STORAGE_PACKAGE_INVALID'
|
|
|
|
|
|
def test_forged_newer_name_does_not_override_manifest_or_roll_back(admitted):
|
|
import shutil
|
|
_,store,head=admitted
|
|
shutil.copyfile(head,head.parent/'g100000000.zip')
|
|
with pytest.raises(LifecycleError) as error:store.latest('case-A','stable-package')
|
|
assert error.value.code=='STORAGE_CONFLICT'
|
|
|
|
|
|
def test_real_process_crash_after_publication(admitted):
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
workspace, _, _ = admitted
|
|
script = "from viano_bug_analysis.lifecycle import Workspace,CheckpointStore,State; import os,sys; s=CheckpointStore(Workspace(sys.argv[1])); s._index=lambda *args: os._exit(23); s.advance('case-A','stable-package',State.BUILDING_CASE,step_id='process-build')"
|
|
env = dict(os.environ, PYTHONPATH=str(Path(__file__).resolve().parents[1] / "src"))
|
|
process = subprocess.run([sys.executable, "-c", script, str(workspace.root)], env=env, capture_output=True, timeout=15)
|
|
assert process.returncode == 23, process.stderr.decode()
|
|
restarted = CheckpointStore(workspace)
|
|
path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="process-build")
|
|
with PackageReader(path) as reader:
|
|
assert reader.manifest["package_generation"] == 2
|
|
assert len(checkpoint_chain(reader)) == 2
|
|
assert len(list(path.parent.glob("g*.zip"))) == 2
|
|
|
|
|
|
def test_reserved_checkpoint_artifacts_rejected(admitted):
|
|
_, store, original = admitted
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", artifacts={CHECKPOINT_POINTER: None})
|
|
assert error.value.code == "RESERVED_ARTIFACT"
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", additions={"history/checkpoints/user.json": b"{}"})
|
|
assert store.latest("case-A", "stable-package") == original
|
|
|
|
|
|
def test_invalid_block_request_types(admitted):
|
|
_, store, _ = admitted
|
|
for reason, retry in [(3, "2026-01-01T00:00:00Z"), ("reason", 4)]:
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason=reason, retry_not_before=retry)
|
|
assert error.value.code == "INVALID_RETRY"
|
|
|
|
|
|
def test_oversized_checkpoint_not_published(make_package, tmp_path):
|
|
from viano_bug_analysis.package import PackageLimits
|
|
store = CheckpointStore(Workspace(tmp_path / "service"), PackageLimits(max_manifest_bytes=4000))
|
|
_, original = store.intake(make_package())
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="large", details={"data": "x" * 5000})
|
|
assert error.value.code == "CHECKPOINT_LIMIT"
|
|
assert store.latest("case-A", "stable-package") == original
|
|
|
|
|
|
def test_empty_array_details_rejected(admitted):
|
|
_, store, original = admitted
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="bad", details=[])
|
|
assert store.latest("case-A", "stable-package") == original
|
|
|
|
|
|
@pytest.mark.parametrize("reason", ["INVALID_LLM_RESPONSES", "RESEARCH_UNAVAILABLE", "AWAITING_TEST_FEEDBACK"])
|
|
def test_manual_repair_retry_is_explicit_scoped_and_preserves_content(admitted,reason):
|
|
_, store, _ = admitted
|
|
blocked = store.advance("case-A", "stable-package", State.BLOCKED, step_id="repair-block",
|
|
reason=reason, retry_not_before="2099-01-01T00:00:00Z")
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.VALIDATED, step_id="automatic-too-early")
|
|
if reason != "INVALID_LLM_RESPONSES":
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.VALIDATED, step_id="not-repairable",manual_retry=True)
|
|
assert store.latest("case-A", "stable-package") == blocked
|
|
return
|
|
with PackageReader(blocked) as reader:
|
|
before = reader.manifest
|
|
resumed = store.advance("case-A", "stable-package", State.VALIDATED, step_id="operator-repaired",manual_retry=True)
|
|
assert store.advance("case-A", "stable-package", State.VALIDATED, step_id="operator-repaired",manual_retry=True) == resumed
|
|
with PackageReader(resumed) as reader:
|
|
assert reader.read_bytes("unknown") == b"keep"
|
|
assert reader.manifest["analysis_revision"] == before["analysis_revision"]
|
|
assert reader.manifest["artifacts"].keys() == before["artifacts"].keys()
|
|
assert reader.manifest["routing"]["retry_not_before"] is None
|
|
assert checkpoint_chain(reader)[0][1]["details"]["manual_retry"] is True
|
|
assert PackageValidator().validate(resumed).valid
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [{"additions":{"other":b"no"}}, {"artifacts":{"case":"other"}}, {"manual_retry":1}])
|
|
def test_manual_repair_retry_cannot_mutate_artifacts(admitted,changes):
|
|
_, store, _ = admitted
|
|
blocked = store.advance("case-A", "stable-package", State.BLOCKED, step_id="block-repair",
|
|
reason="INVALID_LLM_RESPONSES",retry_not_before="2099-01-01T00:00:00Z")
|
|
with pytest.raises(LifecycleError):
|
|
store.advance("case-A", "stable-package", State.VALIDATED, step_id="bad-repair",
|
|
**{**{"manual_retry":True},**changes})
|
|
assert store.latest("case-A", "stable-package") == blocked
|