203 lines
10 KiB
Python
203 lines
10 KiB
Python
import asyncio
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.trial import drive, main, preflight, TrialError
|
|
from viano_bug_analysis.llm import ProviderUnavailable
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.budget import BudgetManager, BudgetConfig
|
|
from viano_bug_analysis.lifecycle import State
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_case_interpreter import gateway, FixtureAdapter
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
from test_investigator import LoopAdapter, CASE
|
|
from test_diagnosis_builder import FinalAdapter
|
|
from test_change_request_builder import RequestAdapter
|
|
|
|
|
|
class PipelineAdapter:
|
|
def __init__(self, confirmed=True):
|
|
self.calls = 0
|
|
self.adapters = {"case": FixtureAdapter(), "hypotheses": ProposalAdapter(),
|
|
"loop": LoopAdapter(), "diagnosis": FinalAdapter(confirmed),
|
|
"request": RequestAdapter()}
|
|
|
|
async def generate(self, config, request, **kwargs):
|
|
self.calls += 1
|
|
phase = {"case_interpretation_v1": "case", "hypotheses_v1": "hypotheses", "hypotheses_v2": "hypotheses", "hypotheses_v3": "hypotheses", "hypotheses_v4": "hypotheses",
|
|
"investigation_v1": "loop", "investigation_v2": "loop", "investigation_v3": "loop", "diagnosis_v1": "diagnosis", "diagnosis_v2": "diagnosis", "diagnosis_v3": "diagnosis", "diagnosis_v4": "diagnosis", "diagnosis_v5": "diagnosis", "diagnosis_v6": "diagnosis", "diagnosis_v7": "diagnosis", "diagnosis_v8": "diagnosis",
|
|
"change_request_v1": "request", "change_request_v2": "request", "change_request_v3": "request", "change_request_v4": "request", "change_request_v5": "request", "change_request_v6": "request", "change_request_v7": "request", "change_request_v8": "request", "change_request_v9": "request"}[request["prompt_version"]]
|
|
adapter = self.adapters[phase]
|
|
return await adapter.generate(config, request, **kwargs)
|
|
|
|
|
|
@pytest.mark.parametrize("confirmed,expected", [(True, "OUTPUT_READY"), (False, "REVIEW")])
|
|
def test_trial_full_investigation_delivery_and_resume(tool_case, confirmed, expected):
|
|
store, original = tool_case
|
|
adapter = PipelineAdapter(confirmed)
|
|
g = gateway(adapter)
|
|
result = asyncio.run(drive(store, g, *CASE))
|
|
assert result["status"] == expected
|
|
exported = result["exported"]
|
|
assert exported and PackageValidator().validate(exported).valid
|
|
with PackageReader(original) as old, PackageReader(exported) as new:
|
|
for name in old.files:
|
|
if name != "manifest.json":
|
|
assert old.read_bytes(name) == new.read_bytes(name)
|
|
assert new.manifest["package_id"] == old.manifest["package_id"]
|
|
assert bool(new.manifest["artifacts"].get("change_request")) == confirmed
|
|
budget = BudgetManager(store).snapshot(*CASE).state
|
|
calls = adapter.calls
|
|
replay = asyncio.run(drive(store, g, *CASE))
|
|
assert replay["exported"] == exported and adapter.calls == calls
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
|
|
|
|
def test_initial_builder_interpretation_and_hypotheses(case_setup):
|
|
store = case_setup[0]
|
|
adapter = PipelineAdapter()
|
|
result = asyncio.run(drive(store, gateway(adapter), *CASE, max_dispatches=3))
|
|
assert result["status"] == "YIELDED"
|
|
assert result["state"] == State.GENERATING_HYPOTHESES
|
|
assert adapter.calls == 2
|
|
|
|
|
|
@pytest.mark.parametrize("phase,dispatches,calls",[("case",1,0),("interpretation",2,1),("interpretation",10,1)])
|
|
def test_stop_after_confirmed_phase_and_safe_repeated_command(case_setup,phase,dispatches,calls):
|
|
store=case_setup[0]
|
|
adapter=PipelineAdapter();g=gateway(adapter)
|
|
result=asyncio.run(drive(store,g,*CASE,max_dispatches=dispatches,stop_after=phase))
|
|
assert result["status"]=="PHASE_COMPLETE" and adapter.calls==calls
|
|
with PackageReader(result["package"]) as reader:
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
replay=asyncio.run(drive(store,g,*CASE,max_dispatches=10,stop_after=phase))
|
|
assert replay["package"]==result["package"] and adapter.calls==calls
|
|
|
|
|
|
def test_retry_returns_without_sleep_and_blocked_budget_kept(tool_case):
|
|
store = tool_case[0]
|
|
adapter = LoopAdapter(unavailable=True)
|
|
g = gateway(adapter)
|
|
g.config = replace(g.config, max_attempts=1)
|
|
result = asyncio.run(drive(store, g, *CASE))
|
|
assert result["status"] == "BLOCKED"
|
|
budget = BudgetManager(store).snapshot(*CASE).state
|
|
calls = len(adapter.contexts)
|
|
assert asyncio.run(drive(store, g, *CASE))["status"] == "BLOCKED"
|
|
assert len(adapter.contexts) == calls
|
|
assert BudgetManager(store).snapshot(*CASE).state == budget
|
|
|
|
|
|
def test_retry_pending_and_early_resume_are_nonblocking(case_setup):
|
|
store = case_setup[0]
|
|
result = asyncio.run(drive(store, gateway(FixtureAdapter(ProviderUnavailable("LLM_CONNECTION"))), *CASE))
|
|
assert result["status"] == "RETRY_PENDING" and result["retry_not_before"]
|
|
result = asyncio.run(drive(store, gateway(FixtureAdapter()), *CASE))
|
|
assert result["status"] == "DEFERRED"
|
|
|
|
|
|
def test_preflight_no_network_no_secret_echo(tmp_path, monkeypatch, capsys):
|
|
config = tmp_path / "llm.toml"
|
|
config.write_text('[llm]\nprivacy_profile="cloud_allowed"\n[[llm.providers]]\nid="test"\nmodel="test"\nprivacy="cloud"\nbase_url="https://example.invalid/v1"\napi_key_env="TRIAL_TEST_KEY"\n')
|
|
monkeypatch.delenv("TRIAL_TEST_KEY", raising=False)
|
|
assert main(["check", "--llm-config", str(config)]) == 2
|
|
assert "LLM_CREDENTIALS" in capsys.readouterr().err
|
|
monkeypatch.setenv("TRIAL_TEST_KEY", "private-value")
|
|
assert main(["check", "--llm-config", str(config)]) == 0
|
|
output = capsys.readouterr().out
|
|
assert json.loads(output)["network_tested"] is False
|
|
assert "private-value" not in output
|
|
|
|
|
|
def test_cli_rejects_ambiguous_identity():
|
|
with pytest.raises(SystemExit):
|
|
main(["run", "--package", "one.zip", "--case-id", "a", "--package-id", "b"])
|
|
|
|
|
|
def test_cli_interpretation_stop_exits_successfully(case_setup,monkeypatch,capsys):
|
|
import viano_bug_analysis.trial as trial
|
|
store=case_setup[0];adapter=PipelineAdapter()
|
|
monkeypatch.setattr(trial,"preflight",lambda path:gateway(adapter))
|
|
command=["run","--workspace",str(store.workspace.root),"--case-id",CASE[0],"--package-id",CASE[1],
|
|
"--stop-after","interpretation","--max-dispatches","6"]
|
|
assert main(command)==0
|
|
assert json.loads(capsys.readouterr().out)["status"]=="PHASE_COMPLETE" and adapter.calls==1
|
|
assert main(command)==0 and adapter.calls==1
|
|
|
|
|
|
def test_cli_run_connects_telemetry_and_publishes(tool_case, monkeypatch, capsys):
|
|
import viano_bug_analysis.trial as trial
|
|
store = tool_case[0]
|
|
monkeypatch.setattr(trial, "preflight", lambda path: gateway(PipelineAdapter()))
|
|
assert main(["run", "--workspace", str(store.workspace.root),
|
|
"--case-id", CASE[0], "--package-id", CASE[1]]) == 0
|
|
captured = capsys.readouterr()
|
|
assert json.loads(captured.out)["status"] == "OUTPUT_READY"
|
|
assert '"state"' in captured.err
|
|
directory = store.workspace.runtime / "telemetry"
|
|
assert (directory / "service.jsonl").is_file()
|
|
assert 'vba_llm_rounds_total' in (directory / "metrics.prom").read_text()
|
|
assert json.loads((directory / "status.json").read_text())["active_operations"] == []
|
|
|
|
|
|
def test_human_test_returns_blocked_with_pending_budget(tool_case):
|
|
store = tool_case[0]
|
|
adapter = LoopAdapter(test=True)
|
|
result = asyncio.run(drive(store, gateway(adapter), *CASE,
|
|
policies={"budget": BudgetConfig(action_costs={"testing.request_human": 50})}))
|
|
assert result["status"] == "BLOCKED"
|
|
assert result["routing"]["reason"] == "AWAITING_TEST_FEEDBACK"
|
|
before = BudgetManager(store).snapshot(*CASE).state
|
|
assert asyncio.run(drive(store, gateway(adapter), *CASE))["status"] == "BLOCKED"
|
|
assert before == BudgetManager(store).snapshot(*CASE).state
|
|
|
|
|
|
def test_dispatch_yield_resumes_without_repeating_first_model_call(tool_case):
|
|
store = tool_case[0]
|
|
adapter = PipelineAdapter()
|
|
g = gateway(adapter)
|
|
first = asyncio.run(drive(store, g, *CASE, max_dispatches=1))
|
|
assert first["status"] == "YIELDED" and adapter.calls == 1
|
|
result = asyncio.run(drive(store, g, *CASE))
|
|
assert result["status"] == "OUTPUT_READY" and adapter.calls == 8
|
|
|
|
|
|
def test_invalid_early_model_reply_reaches_bounded_review(case_setup):
|
|
from test_case_interpreter import payload
|
|
adapter = FixtureAdapter({**payload(), "unexpected": "invalid"})
|
|
result = asyncio.run(drive(case_setup[0], gateway(adapter), *CASE))
|
|
assert result["status"] == "REVIEW"
|
|
assert len(adapter.requests) == 3
|
|
|
|
|
|
def test_cli_corrupt_inbox_is_quarantined_without_model_calls(tmp_path, monkeypatch, capsys):
|
|
import viano_bug_analysis.trial as trial
|
|
from viano_bug_analysis.lifecycle import Workspace
|
|
workspace = Workspace(tmp_path / "work")
|
|
source = workspace.inbox / "arbitrary.bin"
|
|
source.write_bytes(b"corrupt zip")
|
|
adapter = PipelineAdapter()
|
|
monkeypatch.setattr(trial, "preflight", lambda path: gateway(adapter))
|
|
assert main(["run", "--workspace", str(workspace.root), "--package", str(source)]) == 2
|
|
assert json.loads(capsys.readouterr().out)["status"] == "QUARANTINED"
|
|
assert adapter.calls == 0
|
|
assert list(workspace.quarantine.glob("*/validation_error.json"))
|
|
|
|
|
|
def test_missing_case_is_safe_cli_error(tmp_path, monkeypatch, capsys):
|
|
import viano_bug_analysis.trial as trial
|
|
monkeypatch.setattr(trial, "preflight", lambda path: gateway(PipelineAdapter()))
|
|
assert main(["run", "--workspace", str(tmp_path / "work"),
|
|
"--case-id", "missing", "--package-id", "missing"]) == 2
|
|
output = capsys.readouterr().err
|
|
assert '"CASE_NOT_FOUND"' in output and "Traceback" not in output
|
|
|
|
|
|
@pytest.mark.parametrize("limit", [0, -1, True])
|
|
def test_dispatch_limit_is_controlled(tool_case, limit):
|
|
with pytest.raises(TrialError, match="TRIAL_DISPATCH_LIMIT"):
|
|
asyncio.run(drive(tool_case[0], gateway(LoopAdapter()), *CASE, max_dispatches=limit))
|