118 lines
7.9 KiB
Python
118 lines
7.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
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.tools.database_query import DatabaseQueryTool
|
|
from viano_bug_analysis.profiles.vcore.db_tools import normalize_schema_descriptor
|
|
code_ref = next(r["path"] for r in case["source_context"]["refs"] if r["path"].endswith(".cpp"))
|
|
schema_ref = next(r["path"] for r in case["database_context"]["refs"] if r["path"].endswith("/database/schemas/quest_template.json"))
|
|
rows_ref = next(r["path"] for r in case["database_context"]["refs"] if r["path"].endswith("/database/quest_template.json"))
|
|
request = {"format":"viano-investigation-request","format_version":1,"case_id":identity[0],"package_id":identity[1],"analysis_revision":1,
|
|
"requests":[{"request_id":"source-query","hypothesis_id":"H1","action_id":"source-query","operation":"source.read_context",
|
|
"arguments":{"source_ref":code_ref,"start_line":1,"end_line":3}},
|
|
{"request_id":"schema-query","hypothesis_id":"H2","action_id":"schema-query","operation":"database.describe",
|
|
"arguments":{"source_ref":schema_ref}},
|
|
{"request_id":"rows-query","hypothesis_id":"H3","action_id":"rows-query","operation":"database.lookup",
|
|
"arguments":{"source_ref":rows_ref,"filters":{"ID":11117},"columns":["ID"],"offset":0,"limit":10}}]}
|
|
before_query_hash = hashlib.sha256(accounted.package.read_bytes()).hexdigest()
|
|
queried = InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor)).query(accounted.package,request)
|
|
assert len(queried["results"][0]["data"]["matches"]) == 3
|
|
assert queried["results"][1]["data"]["table"] == "quest_template"
|
|
assert queried["results"][1]["data"]["coverage"] == "schema_only"
|
|
assert queried["results"][2]["data"]["matches"][0]["row"] == {"ID":11117}
|
|
assert queried["results"][2]["data"]["coverage"] == "unknown"
|
|
assert hashlib.sha256(accounted.package.read_bytes()).hexdigest() == before_query_hash
|
|
assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] == original_hashes
|