167 lines
8.7 KiB
Python
167 lines
8.7 KiB
Python
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)
|