385 lines
21 KiB
Python
385 lines
21 KiB
Python
import json
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from importlib.resources import files
|
|
|
|
from viano_bug_analysis.case import CaseBuilder, CaseAssembler, CaseBuilderLimits, CaseBuildError, render_summary
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, LifecycleError, State, Workspace
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter
|
|
|
|
|
|
@pytest.fixture
|
|
def case_setup(manifest, make_package, tmp_path):
|
|
manifest["artifacts"]["case_inputs"] = "arbitrary/index.data"
|
|
report = {"format": "viano-bug-report", "format_version": 1, "title": "Generic problem", "summary": "Reported behavior",
|
|
"observations": [{"text": "The switch remains off", "author": "Tester A", "observed_at": "2026-01-01T01:00:00Z"}],
|
|
"comments": [{"text": "Seen twice"}], "tester_hypotheses": [{"text": "Perhaps a timing issue"}],
|
|
"expected_behavior": [{"text": "The switch turns on"}], "expected_behavior_complete": True,
|
|
"entities": [{"type": "component", "id": 42, "roles": ["focus"]}]}
|
|
reproduction = {"format": "viano-reproduction", "format_version": 1, "steps": ["Start", "Press switch"],
|
|
"observations": [{"text": "Still off"}], "events_ref": "raw/events", "attachments": ["raw/binary"],
|
|
"facts": [{"name": "attempt", "value": 2}], "entities": [{"type": "component", "id": "42", "roles": ["observed"]}]}
|
|
index = {"format": "viano-case-inputs", "format_version": 1, "reports": ["anything/report.data"],
|
|
"reproductions": ["anything/repro.data"], "context": {"source": ["raw/source"], "database": ["raw/db"], "runtime": ["raw/runtime"], "research": ["raw/research"]}}
|
|
entries = {"arbitrary/index.data": json.dumps(index), "anything/report.data": json.dumps(report), "anything/repro.data": json.dumps(reproduction),
|
|
"raw/events": b"many events", "raw/binary": b"\x00\xff", "raw/source": b"source raw", "raw/db": b"db raw", "raw/runtime": b"runtime raw", "raw/research": b"supplied external notes", "unknown/vendor": b"opaque"}
|
|
source = make_package(entries=entries)
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(source)
|
|
return store, CaseBuilder(store), source, entries, report, index
|
|
|
|
|
|
@pytest.mark.parametrize("contract", ["case", "case-inputs", "report", "reproduction"])
|
|
def test_contract_schemas_valid(contract):
|
|
schema = json.loads(files("viano_bug_analysis.schemas").joinpath(f"{contract}-v1.schema.json").read_text())
|
|
Draft202012Validator.check_schema(schema)
|
|
|
|
|
|
def test_full_build_preserves_separation_and_provenance(case_setup):
|
|
store, builder, source, entries, _, _ = case_setup
|
|
original = source.read_bytes()
|
|
result = builder.build("case-A", "stable-package")
|
|
assert not result.reused
|
|
assert result.analysis_revision == 1
|
|
assert PackageValidator().validate(result.package).valid
|
|
assert source.read_bytes() == original
|
|
with PackageReader(result.package) as reader:
|
|
manifest = reader.manifest
|
|
assert manifest["state"] == State.BUILDING_CASE
|
|
assert manifest["package_generation"] == 3
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
assert case["case"]["source_generation"] == 2
|
|
assert case["human_observations"]["observations"][0]["text"] == "The switch remains off"
|
|
assert len(case["human_observations"]["observations"]) == 2
|
|
assert case["human_observations"]["tester_hypotheses"][0]["text"] == "Perhaps a timing issue"
|
|
assert case["expected_behavior"]["completeness"] == "reported_complete"
|
|
assert case["human_observations"]["observations"][0]["provenance"] == {"kind": "tester", "source_ref": "anything/report.data", "json_pointer": "/observations/0", "author": "Tester A", "observed_at": "2026-01-01T01:00:00Z"}
|
|
assert case["runtime_facts"]["facts"][0]["id"] == "F0001"
|
|
assert case["runtime_facts"]["facts"][0]["provenance"]["json_pointer"] == "/facts/0"
|
|
assert case["entities"][0]["id"] == "42"
|
|
assert case["entities"][0]["roles"] == ["focus", "observed"]
|
|
assert len(case["entities"][0]["provenance"]) == 2
|
|
assert case["research"]["performed"] is False
|
|
assert [item["path"] for item in case["inputs"]["unclassified"]] == ["unknown/vendor"]
|
|
assert reader.read_bytes(result.summary_ref).decode() == render_summary(case)
|
|
for path, value in entries.items():
|
|
assert reader.read_bytes(path) == (value.encode() if isinstance(value, str) else value)
|
|
|
|
|
|
def test_completed_build_is_idempotent(case_setup):
|
|
store, builder, _, _, _, _ = case_setup
|
|
first = builder.build("case-A", "stable-package")
|
|
again = CaseBuilder(CheckpointStore(store.workspace)).build("case-A", "stable-package")
|
|
assert again.reused and again.package == first.package
|
|
assert len(list(first.package.parent.glob("g*.zip"))) == 3
|
|
|
|
|
|
def test_missing_index_marks_gaps_without_guessing(make_package, tmp_path):
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(make_package(entries={"original/bug_report.json": b"not a declared report"}))
|
|
result = CaseBuilder(store).build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
codes = {issue["code"] for issue in case["data_quality"]["issues"]}
|
|
assert {"MISSING_INPUT_INDEX", "MISSING_REPORT", "MISSING_OBSERVATIONS", "MISSING_REPRODUCTION", "EXPECTED_BEHAVIOR_MISSING"} <= codes
|
|
assert case["bug"]["reports"] == []
|
|
assert case["inputs"]["unclassified"][0]["path"] == "original/bug_report.json"
|
|
|
|
|
|
def rewrite_input(case_setup, tmp_path, changes, updates=None):
|
|
store = case_setup[0]
|
|
current = store.latest("case-A", "stable-package")
|
|
incoming = tmp_path / "updated-input"
|
|
PackageWriter().write(current, incoming, additions={name: json.dumps(data).encode() for name, data in changes.items()}, manifest_updates=updates)
|
|
store.intake(incoming)
|
|
return current
|
|
|
|
|
|
def test_reopen_inherits_previous_reports_and_facts(case_setup, tmp_path):
|
|
store, builder, _, _, _, _ = case_setup
|
|
first = builder.build("case-A", "stable-package")
|
|
for index, target in enumerate([State.INTERPRETING_CASE, State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES, State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF, State.REOPENED]):
|
|
store.advance("case-A", "stable-package", target, step_id=f"test-stage-{index}")
|
|
report = {"format": "viano-bug-report", "format_version": 1, "observations": [{"text": "New feedback"}]}
|
|
index = {"format": "viano-case-inputs", "format_version": 1, "reports": ["feedback/new-report"]}
|
|
rewrite_input(case_setup, tmp_path, {"feedback/new-report": report, "inputs/new-index": index}, {"artifacts": {"case_inputs": "inputs/new-index"}})
|
|
result = builder.build("case-A", "stable-package")
|
|
assert result.analysis_revision == 2
|
|
with PackageReader(result.package) as reader:
|
|
new = json.loads(reader.read_bytes(result.case_ref))
|
|
old = json.loads(reader.read_bytes(first.case_ref))
|
|
assert new["human_observations"]["observations"][:2] == old["human_observations"]["observations"]
|
|
assert new["human_observations"]["observations"][-1]["text"] == "New feedback"
|
|
assert new["runtime_facts"]["facts"] == old["runtime_facts"]["facts"]
|
|
assert new["inputs"]["prior_cases"][0]["path"] == first.case_ref
|
|
assert reader.read_bytes(first.summary_ref) == render_summary(old).encode()
|
|
|
|
|
|
def test_new_data_after_completion_requires_reopen(case_setup, tmp_path):
|
|
_, builder, _, _, _, _ = case_setup
|
|
builder.build("case-A", "stable-package")
|
|
rewrite_input(case_setup, tmp_path, {"feedback/new": {"anything": "new"}})
|
|
with pytest.raises(CaseBuildError) as error:
|
|
builder.build("case-A", "stable-package")
|
|
assert error.value.code == "STALE_CASE"
|
|
|
|
|
|
def test_completion_crash_reuses_output(case_setup, monkeypatch):
|
|
store, builder, _, _, _, _ = case_setup
|
|
original_index = store._index
|
|
calls = 0
|
|
def crash(*args):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
raise LifecycleError("SIMULATED_CRASH", "after case publication")
|
|
original_index(*args)
|
|
monkeypatch.setattr(store, "_index", crash)
|
|
with pytest.raises(CaseBuildError):
|
|
builder.build("case-A", "stable-package")
|
|
result = CaseBuilder(CheckpointStore(store.workspace)).build("case-A", "stable-package")
|
|
assert result.reused
|
|
assert len(list(result.package.parent.glob("g*.zip"))) == 3
|
|
|
|
|
|
def test_unconfirmed_build_retries_without_duplicate_start(case_setup, monkeypatch):
|
|
store, builder, _, _, _, _ = case_setup
|
|
original_write = store.writer.write
|
|
calls = 0
|
|
def crash(*args, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
raise LifecycleError("SIMULATED_CRASH", "before case publication")
|
|
return original_write(*args, **kwargs)
|
|
monkeypatch.setattr(store.writer, "write", crash)
|
|
with pytest.raises(CaseBuildError):
|
|
builder.build("case-A", "stable-package")
|
|
result = CaseBuilder(CheckpointStore(store.workspace)).build("case-A", "stable-package")
|
|
assert not result.reused
|
|
assert len(list(result.package.parent.glob("g*.zip"))) == 3
|
|
|
|
|
|
def test_portable_completed_case(case_setup, tmp_path):
|
|
_, builder, _, _, _, _ = case_setup
|
|
first = builder.build("case-A", "stable-package")
|
|
other = CheckpointStore(Workspace(tmp_path / "other-service"))
|
|
other.intake(first.package)
|
|
result = CaseBuilder(other).build("case-A", "stable-package")
|
|
assert result.reused
|
|
assert result.package == other.latest("case-A", "stable-package")
|
|
|
|
|
|
def test_summary_deterministic_and_escapes_untrusted_text(case_setup):
|
|
_, builder, _, _, _, _ = case_setup
|
|
result = builder.build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
case["human_observations"]["observations"][0]["text"] = "<script>alert(1)</script> [link](javascript:bad)\n# fake heading"
|
|
summary = render_summary(case)
|
|
assert summary == render_summary(case)
|
|
assert "<script>" not in summary
|
|
assert "[link](javascript:" not in summary
|
|
assert "\n# fake heading" not in summary
|
|
|
|
|
|
@pytest.mark.parametrize("limit,value,code", [("max_documents", 1, "STRUCTURED_INPUT_LIMIT"), ("max_structured_bytes", 50, "STRUCTURED_INPUT_LIMIT"), ("max_json_bytes", 10, "INPUT_JSON"), ("max_case_bytes", 50, "CASE_LIMIT"), ("max_summary_bytes", 10, "SUMMARY_LIMIT")])
|
|
def test_operational_limits(case_setup, limit, value, code):
|
|
store = case_setup[0]
|
|
builder = CaseBuilder(store, CaseBuilderLimits(**{limit: value}))
|
|
with pytest.raises(CaseBuildError) as error:
|
|
builder.build("case-A", "stable-package")
|
|
assert error.value.code == code
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
assert reader.manifest["artifacts"]["case"] is None
|
|
|
|
|
|
def test_concurrent_change_rejects_prepared_case(case_setup, monkeypatch):
|
|
store, builder, _, _, _, _ = case_setup
|
|
original = builder.assembler.assemble
|
|
def change(reader):
|
|
case = original(reader)
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="concurrent-work", additions={"new-data": b"new"})
|
|
return case
|
|
monkeypatch.setattr(builder.assembler, "assemble", change)
|
|
with pytest.raises(CaseBuildError) as error:
|
|
builder.build("case-A", "stable-package")
|
|
assert error.value.code == "GENERATION_CHANGED"
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
assert reader.manifest["artifacts"]["case"] is None
|
|
|
|
|
|
def test_expected_generation_check_and_successful_replay(case_setup):
|
|
store = case_setup[0]
|
|
with pytest.raises(LifecycleError) as error:
|
|
store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="prepared", expected_generation=0)
|
|
assert error.value.code == "GENERATION_CHANGED"
|
|
first = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="prepared", expected_generation=1)
|
|
assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="prepared", expected_generation=1) == first
|
|
|
|
|
|
@pytest.mark.parametrize("change,code", [("version", "INPUT_SCHEMA"), ("path", "INPUT_REFERENCE"), ("missing", "INPUT_REFERENCE"), ("duplicate", "INPUT_DUPLICATE"), ("report", "INPUT_SCHEMA"), ("json", "INPUT_JSON")])
|
|
def test_invalid_declared_input_is_controlled(change, code, manifest, make_package, tmp_path):
|
|
manifest["artifacts"]["case_inputs"] = "index"
|
|
index = {"format": "viano-case-inputs", "format_version": 1, "reports": ["report"]}
|
|
report = {"format": "viano-bug-report", "format_version": 1, "observations": [{"text": "observed"}]}
|
|
if change == "version":
|
|
index["format_version"] = 2
|
|
elif change == "path":
|
|
index["reports"] = ["../report"]
|
|
elif change == "missing":
|
|
index["reports"] = ["absent"]
|
|
elif change == "duplicate":
|
|
index["reports"] = ["report", "report"]
|
|
elif change == "report":
|
|
report["observations"] = "not a list"
|
|
entries = {"index": json.dumps(index), "report": json.dumps(report) if change != "json" else '{"format":"viano-bug-report","format":"duplicate"}'}
|
|
original = make_package(entries=entries)
|
|
before = original.read_bytes()
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(original)
|
|
with pytest.raises(CaseBuildError) as error:
|
|
CaseBuilder(store).build("case-A", "stable-package")
|
|
assert error.value.code == code
|
|
assert original.read_bytes() == before
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
assert reader.manifest["artifacts"]["case"] is None
|
|
|
|
|
|
def test_explicit_conflicts_are_traced_without_semantic_guessing(manifest, make_package, tmp_path):
|
|
manifest["artifacts"]["case_inputs"] = "index"
|
|
reports = [
|
|
{"format": "viano-bug-report", "format_version": 1, "observations": [{"text": "Open"}], "expected_behavior": [{"text": "Should be open", "claim": {"key": "switch_state", "value": "open"}}], "expected_behavior_complete": True},
|
|
{"format": "viano-bug-report", "format_version": 1, "observations": [{"text": "Closed"}], "expected_behavior": [{"text": "Should be closed", "claim": {"key": "switch_state", "value": "closed"}}], "expected_behavior_complete": False},
|
|
]
|
|
index = {"format": "viano-case-inputs", "format_version": 1, "reports": ["a", "b"]}
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(make_package(entries={"index": json.dumps(index), "a": json.dumps(reports[0]), "b": json.dumps(reports[1])}))
|
|
result = CaseBuilder(store).build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
conflicts = [issue for issue in case["data_quality"]["issues"] if issue["code"] == "CONFLICTING_CLAIMS"]
|
|
assert len(conflicts) == 1
|
|
assert "expected_behavior" in conflicts[0]["message"]
|
|
assert conflicts[0]["refs"] == [{"source_ref": "a", "json_pointer": "/expected_behavior/0"}, {"source_ref": "b", "json_pointer": "/expected_behavior/0"}]
|
|
assert case["expected_behavior"]["completeness"] == "conflicting"
|
|
assert len(case["human_observations"]["observations"]) == 2
|
|
|
|
|
|
def test_plain_language_disagreement_is_not_interpreted(manifest, make_package, tmp_path):
|
|
manifest["artifacts"]["case_inputs"] = "index"
|
|
report = {"format": "viano-bug-report", "format_version": 1, "observations": [{"text": "The light is on"}, {"text": "The light is off"}], "tester_hypotheses": [{"text": "It must be broken"}]}
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(make_package(entries={"index": json.dumps({"format": "viano-case-inputs", "format_version": 1, "reports": ["report"]}), "report": json.dumps(report)}))
|
|
result = CaseBuilder(store).build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
assert not any(issue["code"] == "CONFLICTING_CLAIMS" for issue in case["data_quality"]["issues"])
|
|
assert case["runtime_facts"]["facts"] == []
|
|
|
|
|
|
def test_deterministic_assembly_for_same_snapshot(case_setup):
|
|
store = case_setup[0]
|
|
path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="start")
|
|
with PackageReader(path) as reader:
|
|
assert CaseAssembler().assemble(reader) == CaseAssembler().assemble(reader)
|
|
|
|
|
|
def test_completed_case_summary_limit_applies_on_reuse(case_setup):
|
|
store, builder, _, _, _, _ = case_setup
|
|
builder.build("case-A", "stable-package")
|
|
with pytest.raises(CaseBuildError) as error:
|
|
CaseBuilder(store, CaseBuilderLimits(max_summary_bytes=10)).build("case-A", "stable-package")
|
|
assert error.value.code == "SUMMARY_LIMIT"
|
|
|
|
|
|
def test_builder_rejects_unknown_case_and_wrong_phase(case_setup):
|
|
store, builder, _, _, _, _ = case_setup
|
|
with pytest.raises(CaseBuildError) as error:
|
|
builder.build("unknown", "unknown")
|
|
assert error.value.code == "CASE_NOT_FOUND"
|
|
store.advance("case-A", "stable-package", State.FAILED_TO_ANALYZE, step_id="unavailable", reason="test")
|
|
with pytest.raises(CaseBuildError) as error:
|
|
builder.build("case-A", "stable-package")
|
|
assert error.value.code == "INVALID_BUILD_STATE"
|
|
|
|
|
|
@pytest.mark.parametrize("value", [0, -1, True, 1.5])
|
|
def test_invalid_builder_limits(value):
|
|
with pytest.raises(ValueError):
|
|
CaseBuilderLimits(max_documents=value)
|
|
|
|
|
|
def test_large_fact_values_remain_referenced(case_setup):
|
|
store = case_setup[0]
|
|
result = CaseBuilder(store, CaseBuilderLimits(max_inline_fact_bytes=1)).build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
fact = case["runtime_facts"]["facts"][0]
|
|
# JSON encoding of the integer 2 occupies one byte, so remains inline.
|
|
assert fact["value"] == 2
|
|
|
|
|
|
def test_large_fact_does_not_embed_raw_array(manifest, make_package, tmp_path):
|
|
manifest["artifacts"]["case_inputs"] = "index"
|
|
payload = list(range(100))
|
|
reproduction = {"format": "viano-reproduction", "format_version": 1, "facts": [{"name": "raw_values", "value": payload}]}
|
|
index = {"format": "viano-case-inputs", "format_version": 1, "reproductions": ["repro"]}
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(make_package(entries={"index": json.dumps(index), "repro": json.dumps(reproduction)}))
|
|
builder = CaseBuilder(store, CaseBuilderLimits(max_inline_fact_bytes=20))
|
|
result = builder.build("case-A", "stable-package")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
fact = case["runtime_facts"]["facts"][0]
|
|
assert "value" not in fact
|
|
assert fact["value_ref"] == {"source_ref": "repro", "json_pointer": "/facts/0/value"}
|
|
assert json.loads(reader.read_bytes("repro"))["facts"][0]["value"] == payload
|
|
assert "referenced at repro /facts/0/value" in reader.read_bytes(result.summary_ref).decode()
|
|
assert builder.build("case-A", "stable-package").reused
|
|
|
|
|
|
def test_real_process_exit_after_case_publication(case_setup):
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
store = case_setup[0]
|
|
script = """
|
|
import os, sys
|
|
from viano_bug_analysis.lifecycle import Workspace, CheckpointStore
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
store = CheckpointStore(Workspace(sys.argv[1]))
|
|
original = store._index
|
|
def terminate(path, manifest):
|
|
if manifest['artifacts'].get('case'):
|
|
os._exit(24)
|
|
original(path, manifest)
|
|
store._index = terminate
|
|
CaseBuilder(store).build('case-A', 'stable-package')
|
|
"""
|
|
env = dict(os.environ, PYTHONPATH=str(Path(__file__).resolve().parents[1] / "src"))
|
|
process = subprocess.run([sys.executable, "-c", script, str(store.workspace.root)], env=env, capture_output=True, timeout=15)
|
|
assert process.returncode == 24, process.stderr.decode()
|
|
result = CaseBuilder(CheckpointStore(store.workspace)).build("case-A", "stable-package")
|
|
assert result.reused
|
|
assert len(list(result.package.parent.glob("g*.zip"))) == 3
|
|
|
|
|
|
def test_start_replay_can_observe_already_completed_build(case_setup, monkeypatch):
|
|
store, builder, _, _, _, _ = case_setup
|
|
original = store.advance
|
|
injected = False
|
|
def complete_elsewhere(*args, **kwargs):
|
|
nonlocal injected
|
|
result = original(*args, **kwargs)
|
|
if kwargs["step_id"].endswith(":start") and not injected:
|
|
injected = True
|
|
return CaseBuilder(CheckpointStore(store.workspace)).build("case-A", "stable-package").package
|
|
return result
|
|
monkeypatch.setattr(store, "advance", complete_elsewhere)
|
|
result = builder.build("case-A", "stable-package")
|
|
assert result.reused
|
|
assert len(list(result.package.parent.glob("g*.zip"))) == 3
|