import hashlib import json import pytest from viano_bug_analysis.case.expected_behavior import ExpectedBehaviorResolver, collect_expected_behavior from viano_bug_analysis.case import CaseAssembler, CaseBuildError from viano_bug_analysis.lifecycle import State from viano_bug_analysis.lifecycle.storage import json_bytes from viano_bug_analysis.package import PackageReader, PackageWriter, PackageValidator from viano_bug_analysis.research import ResearchError, ResearchLimits, ResearchService from test_case_builder import case_setup def request(): return ExpectedBehaviorResolver().request( {"case_id": "case-A", "package_id": "stable-package", "analysis_revision": 1, "project": {}}, [], [], {"statements": [], "completeness": "missing", "provenance": "none"}) def result(req, status="complete"): return {"format": "viano-research-result", "format_version": 1, "request_hash": hashlib.sha256(json_bytes(req)).hexdigest(), "provider": "fixture", "scope": "intended_behavior_only", "status": status, "sources": [{"url": "https://example.org/manual", "title": "Manual", "retrieved_at": "2026-01-01T00:00:00Z", "excerpt": "The switch turns on."}], "statements": [] if status == "not_found" else [{"text": "The switch turns on.", "source_indices": [0]}]} class FixtureProvider: def __init__(self, response): self.response = response def lookup(self, request, *, timeout_seconds): assert timeout_seconds > 0 return self.response @pytest.mark.parametrize("flag,status", [(True,"reported_complete"),(False,"reported_partial"),(None,"unspecified")]) def test_explicit_completeness(flag, status): expected = collect_expected_behavior([("report", {"expected_behavior": [{"text":"Expected"}], "expected_behavior_complete": flag})]) assert expected["completeness"] == status req = ExpectedBehaviorResolver().request({"case_id":"c","package_id":"p","analysis_revision":1,"project":{}}, [], [], expected) assert (req is None) == (flag is True) @pytest.mark.parametrize("status,completeness", [("complete","researched_complete"),("partial","researched_partial"),("not_found","missing")]) def test_resolution(status, completeness): expected = {"statements":[],"completeness":"missing","provenance":"none"} resolved = ExpectedBehaviorResolver().resolve(expected, [("research/result.json",result(request(), status))]) assert resolved["completeness"] == completeness assert expected["statements"] == [] if resolved["statements"]: assert resolved["statements"][0]["external_sources"][0]["url"].startswith("https:") def test_combined_does_not_override_tester_or_conflict(): expected = collect_expected_behavior([("r", {"expected_behavior":[{"text":"Tester statement"}],"expected_behavior_complete":False})]) resolved = ExpectedBehaviorResolver().resolve(expected, [("research/result.json",result(request()))]) assert resolved["provenance"] == "combined" assert resolved["statements"][0] == expected["statements"][0] expected["completeness"] = "conflicting" assert ExpectedBehaviorResolver().resolve(expected, [("r",result(request()))])["completeness"] == "conflicting" @pytest.mark.parametrize("mutation", ["identity","source","scope","time"]) def test_invalid_provider_is_rejected(mutation): response = result(request()) if mutation == "identity": response["request_hash"] = "0" * 64 if mutation == "source": response["statements"][0]["source_indices"] = [2] if mutation == "scope": response["scope"] = "local_implementation" if mutation == "time": response["sources"][0]["retrieved_at"] = "2026-01-01" with pytest.raises(CaseBuildError): ResearchService(FixtureProvider(response)).collect(request()) def test_provider_failure_controlled(): class Offline: def lookup(self, request, *, timeout_seconds): raise TimeoutError() with pytest.raises(ResearchError, match="temporarily unavailable"): ResearchService(Offline()).collect(request()) with pytest.raises(ResearchError): ResearchService(FixtureProvider(result(request())), ResearchLimits(max_result_bytes=1)).collect(request()) def test_collection_deterministic(): service = ResearchService(FixtureProvider(result(request()))) assert service.collect(request()) == service.collect(request()) def test_package_roundtrip_and_restart(case_setup, tmp_path): store, builder, _, entries, _, index = case_setup active = store.advance("case-A","stable-package",State.BUILDING_CASE,step_id="prepare-research") with PackageReader(active) as reader: case = CaseAssembler().assemble(reader) # Research can also be explicitly supplied to corroborate a complete tester report. req = request() req["reports"] = [{"title":"Generic problem","summary":"Reported behavior"}] req["entities"] = [{"type":"component","id":"42"}] req["tester_expectations"] = ["The switch turns on"] req["project"] = reader.manifest["project"] artifacts = reader.manifest["artifacts"] ref, additions = ResearchService(FixtureProvider(result(req))).collect(req) index = dict(index, expected_research=[ref]) additions["research/new-index.json"] = json_bytes(index) enriched = tmp_path / "enriched.zip" PackageWriter().write(active,enriched,additions=additions,manifest_updates={"artifacts": {**artifacts,"case_inputs":"research/new-index.json"}}) store.intake(enriched) built = builder.build("case-A","stable-package") assert PackageValidator().validate(built.package).valid with PackageReader(built.package) as reader: canonical = json.loads(reader.read_bytes(built.case_ref)) assert canonical["expected_behavior"]["provenance"] == "combined" assert canonical["research"]["performed"] is True for path, value in entries.items(): assert reader.read_bytes(path) == (value.encode() if isinstance(value,str) else value) assert builder.build("case-A","stable-package").reused def test_service_enrichment_missing_expectations(manifest, make_package, tmp_path): from viano_bug_analysis.lifecycle import CheckpointStore, Workspace from viano_bug_analysis.case import CaseBuilder manifest["artifacts"]["case_inputs"] = "input.json" source = make_package(entries={"input.json": json_bytes({"format":"viano-case-inputs","format_version":1,"reports":["report.json"]}), "report.json": json_bytes({"format":"viano-bug-report","format_version":1,"title":"Switch remains off"}), "unknown.bin": b"preserve"}) store = CheckpointStore(Workspace(tmp_path / "workspace")) store.intake(source) active = store.advance("case-A","stable-package",State.BUILDING_CASE,step_id="research-start") class DynamicProvider: calls = 0 def lookup(self, request, *, timeout_seconds): self.calls += 1 assert "source_context" not in request and "database_context" not in request return result(request) provider = DynamicProvider() service = ResearchService(provider) destination = tmp_path / "researched.zip" service.enrich(active, destination) assert provider.calls == 1 assert PackageValidator().validate(destination).valid with PackageReader(active) as before, PackageReader(destination) as after: assert after.manifest["package_generation"] == before.manifest["package_generation"] + 1 assert after.manifest["package_id"] == before.manifest["package_id"] for path in before.files: if path != "manifest.json": assert before.read_bytes(path) == after.read_bytes(path) store.intake(destination) built = CaseBuilder(store).build("case-A","stable-package") with PackageReader(built.package) as reader: case = json.loads(reader.read_bytes(built.case_ref)) assert case["expected_behavior"]["completeness"] == "researched_complete" assert case["expected_behavior"]["provenance"] == "external_research" assert case["runtime_facts"]["facts"] == [] with pytest.raises(ResearchError, match="Reopen"): service.enrich(built.package, tmp_path / "invalid.zip") assert provider.calls == 1 @pytest.mark.parametrize("value", [0, -1, True, float("inf"), float("nan")]) def test_invalid_timeout(value): with pytest.raises(ValueError): ResearchLimits(timeout_seconds=value) @pytest.mark.parametrize("contract", ["research-request", "research-result"]) def test_research_schemas(contract): from importlib.resources import files from jsonschema import Draft202012Validator schema = json.loads(files("viano_bug_analysis.schemas").joinpath(contract + "-v1.schema.json").read_text()) Draft202012Validator.check_schema(schema)