97 lines
5.9 KiB
Python
97 lines
5.9 KiB
Python
"""Opt-in full-archive regression, without machine-specific paths or binary fixtures."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.profiles.vcore import ArchiveReader, VcoreImporter
|
|
from viano_bug_analysis.analysis import CaseInterpreter, HypothesisBuilder
|
|
from test_case_interpreter import FixtureAdapter, gateway, missing_payload
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
import asyncio
|
|
|
|
|
|
@pytest.mark.skipif(not os.environ.get("VBA_WOWDIAG_SAMPLE") or not os.environ.get("VBA_WOWSRC_SAMPLE"),
|
|
reason="Set VBA_WOWDIAG_SAMPLE and VBA_WOWSRC_SAMPLE for the BUG-20 full archive regression")
|
|
def test_full_bug20_package_pair(tmp_path):
|
|
diag, src = Path(os.environ["VBA_WOWDIAG_SAMPLE"]), Path(os.environ["VBA_WOWSRC_SAMPLE"])
|
|
original_hashes = [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)]
|
|
package = tmp_path / "anything.bin"
|
|
VcoreImporter().create(diag, package, package_id="regression-package", case_id="regression-case", wowsrc=src)
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(package)
|
|
result = CaseBuilder(store).build("regression-case", "regression-package")
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
assert len(case["human_observations"]["observations"]) == 3
|
|
assert case["expected_behavior"]["completeness"] == "missing"
|
|
assert case["human_observations"]["tester_hypotheses"] == []
|
|
assert len(case["database_context"]["refs"]) == 103
|
|
assert len(case["source_context"]["refs"]) == 93
|
|
for input_path in (diag, src):
|
|
with ArchiveReader(input_path) as source:
|
|
for member in source.members:
|
|
matching = [path for path in reader.files if path.endswith("/members/" + member)]
|
|
assert len(matching) == 1
|
|
assert reader.read_bytes(matching[0]) == source.read_member(member)
|
|
assert CaseBuilder(store).build("regression-case", "regression-package").reused
|
|
# Offline phase integration only: this fixture is not a test of model reasoning.
|
|
response = missing_payload()
|
|
response["observed_behavior"][0]["text"] = case["human_observations"]["observations"][0]["text"]
|
|
adapter = FixtureAdapter(response)
|
|
interpreter = CaseInterpreter(store, gateway(adapter))
|
|
interpreted = asyncio.run(interpreter.interpret("regression-case", "regression-package"))
|
|
assert interpreted.status == "COMPLETED"
|
|
with PackageReader(result.package) as before, PackageReader(interpreted.package) as after:
|
|
document = json.loads(after.read_bytes(interpreted.interpretation_ref))
|
|
assert document["expected_behavior_completeness"] == "missing"
|
|
assert document["interpretation"]["expected_behavior"] == []
|
|
assert document["interpretation"]["behavior_gap"] is None
|
|
assert after.manifest["artifacts"]["diagnosis"] is None
|
|
assert after.manifest["artifacts"]["change_request"] is None
|
|
for name in before.files:
|
|
if name != "manifest.json":
|
|
assert before.read_bytes(name) == after.read_bytes(name)
|
|
assert PackageValidator().validate(interpreted.package).valid
|
|
assert asyncio.run(interpreter.interpret("regression-case", "regression-package")).reused
|
|
assert len(adapter.requests) == 1
|
|
hypotheses_adapter = ProposalAdapter()
|
|
generated = asyncio.run(HypothesisBuilder(store, gateway(hypotheses_adapter)).build("regression-case", "regression-package"))
|
|
assert generated.status == "COMPLETED"
|
|
assert PackageValidator().validate(generated.package).valid
|
|
with PackageReader(interpreted.package) as before, PackageReader(generated.package) as after:
|
|
hypotheses = json.loads(after.read_bytes(generated.hypotheses_ref))
|
|
registry = json.loads(after.read_bytes(generated.registry_ref))
|
|
assert len(hypotheses["hypotheses"]) == 3
|
|
assert [record["budget_allocation"]["initial"] for record in hypotheses["hypotheses"]] == [100, 100, 100]
|
|
assert registry["created_count"] == 3 and registry["max_hypotheses"] == 5
|
|
assert after.manifest["artifacts"]["diagnosis"] is None
|
|
assert after.manifest["artifacts"]["change_request"] is None
|
|
for name in before.files:
|
|
if name != "manifest.json":
|
|
assert before.read_bytes(name) == after.read_bytes(name)
|
|
assert asyncio.run(HypothesisBuilder(store, gateway(hypotheses_adapter)).build("regression-case", "regression-package")).reused
|
|
assert len(hypotheses_adapter.requests) == 1
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
budget = BudgetManager(store)
|
|
identity = ("regression-case", "regression-package")
|
|
budget.initialize(*identity)
|
|
budget.reserve(*identity, "H1", "offline-query", "source.read_context")
|
|
source_ref = case["source_context"]["refs"][0]["path"]
|
|
budget.record_result(*identity, "H1", "offline-query", {"fixture_observation":"Stored context is available"}, source_refs=[source_ref])
|
|
accounted = budget.settle(*identity, "H1", "offline-query", semantic_score=50, reason="Simulated assessment, no diagnosis")
|
|
assert accounted.state["accounts"]["H1"]["remaining"] == 94
|
|
assert PackageValidator().validate(accounted.package).valid
|
|
with PackageReader(generated.package) as before, PackageReader(accounted.package) as after:
|
|
for name in before.files:
|
|
if name != "manifest.json":
|
|
assert before.read_bytes(name) == after.read_bytes(name)
|
|
assert CaseBuilder(store).build(*identity).reused
|
|
assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] == original_hashes
|