473 lines
24 KiB
Python
473 lines
24 KiB
Python
import asyncio
|
|
import json
|
|
import threading
|
|
from dataclasses import replace
|
|
from datetime import datetime, timedelta, timezone
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, InboxProcessor, State, Workspace
|
|
from viano_bug_analysis.lifecycle.errors import LockBusy
|
|
from viano_bug_analysis.llm import (CaseLLMRunner, GatewayConfig, GatewayUnavailable, LLMError,
|
|
LLMGateway, PromptBuilder, ProviderConfig, ProviderResponse, ProviderUnavailable)
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
|
|
SCHEMA = {"type": "object", "properties": {"answer": {"type": "string"}}, "required": ["answer"], "additionalProperties": False}
|
|
|
|
|
|
def request(privacy="local_only"):
|
|
return PromptBuilder().build(task="Transport test", data={"observation":"Off"}, output_schema=SCHEMA, privacy=privacy)
|
|
|
|
|
|
def provider(name="first", **kw):
|
|
return ProviderConfig(id=name, model="fixture-model", base_url="http://127.0.0.1:8000/v1", adapter="fixture", **kw)
|
|
|
|
|
|
class Adapter:
|
|
def __init__(self, results):
|
|
self.results = results
|
|
self.calls = []
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.calls.append(config.id)
|
|
response = self.results[config.id]
|
|
if isinstance(response, Exception):
|
|
raise response
|
|
return response
|
|
|
|
|
|
def success():
|
|
return ProviderResponse({"answer":"ok"}, "fixture-model", 10, 20)
|
|
|
|
|
|
def gateway(adapter, **kw):
|
|
config = GatewayConfig((provider(), provider("second")), **kw)
|
|
return LLMGateway(config, {"fixture":adapter})
|
|
|
|
|
|
def test_ordered_failover_and_usage():
|
|
adapter = Adapter({"first":ProviderUnavailable("LLM_CONNECTION"),"second":success()})
|
|
config = GatewayConfig((provider(), replace(provider("second"), input_cost_per_million=2, output_cost_per_million=4)))
|
|
output = asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request()))
|
|
assert adapter.calls == ["first","second"]
|
|
assert output["document"] == {"answer":"ok"}
|
|
assert output["metadata"]["input_tokens"] == 10
|
|
assert output["metadata"]["estimated_cost"] == pytest.approx(0.0001)
|
|
assert output["metadata"]["attempts"] == [{"provider":"first","code":"LLM_CONNECTION"}]
|
|
|
|
|
|
def test_first_success_stops():
|
|
adapter = Adapter({"first":success()})
|
|
asyncio.run(gateway(adapter).complete_round(request()))
|
|
assert adapter.calls == ["first"]
|
|
|
|
|
|
def test_all_unavailable_one_round():
|
|
adapter = Adapter({"first":ProviderUnavailable("LLM_CONNECTION"),"second":ProviderUnavailable("LLM_TIMEOUT")})
|
|
with pytest.raises(GatewayUnavailable) as caught:
|
|
asyncio.run(gateway(adapter).complete_round(request()))
|
|
assert len(caught.value.attempts) == 2
|
|
assert adapter.calls == ["first","second"]
|
|
|
|
|
|
def test_cloud_requires_explicit_privacy():
|
|
adapter = Adapter({"cloud":success()})
|
|
config = GatewayConfig((provider(), replace(provider("cloud"), privacy="cloud", base_url="https://cloud.example/v1")), privacy_profile="cloud_allowed")
|
|
adapter.results["first"] = ProviderUnavailable("LLM_CONNECTION")
|
|
with pytest.raises(GatewayUnavailable):
|
|
asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request()))
|
|
assert adapter.calls == ["first"]
|
|
output = asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request("cloud_allowed")))
|
|
assert output["metadata"]["provider"] == "cloud"
|
|
|
|
|
|
@pytest.mark.parametrize("response,code", [(ProviderResponse({"wrong":1},"model"),"LLM_OUTPUT_SCHEMA"),
|
|
(LLMError("LLM_CREDENTIALS"),"LLM_CREDENTIALS"), (RuntimeError("secret"),"LLM_PROVIDER_ERROR"),
|
|
(ProviderResponse({"answer":"ok"},"model",True,1),"LLM_RESPONSE")])
|
|
def test_invalid_output_not_treated_as_unavailability(response, code):
|
|
adapter = Adapter({"first":response,"second":success()})
|
|
with pytest.raises(LLMError) as caught:
|
|
asyncio.run(gateway(adapter).complete_round(request()))
|
|
assert caught.value.code == code
|
|
assert "secret" not in str(caught.value)
|
|
assert adapter.calls == ["first"]
|
|
|
|
|
|
def test_async_deadline_moves_to_next_provider():
|
|
class Slow(Adapter):
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.calls.append(config.id)
|
|
if config.id == "first": await asyncio.sleep(10)
|
|
return success()
|
|
adapter = Slow({})
|
|
config = GatewayConfig((provider(timeout_seconds=0.01),provider("second")))
|
|
output = asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request()))
|
|
assert output["metadata"]["attempts"][0]["code"] == "LLM_TIMEOUT"
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("timeout_seconds",0),("timeout_seconds",float("nan")),("enabled",1),
|
|
("api_key_env","inline secret"),("model","\n"),("base_url","https://user:secret@server.example/v1"),
|
|
("base_url","http://remote.example/v1"),("privacy","unknown"),("response_format","unsupported"),
|
|
("input_cost_per_million",-1),("max_output_tokens",True)])
|
|
def test_invalid_provider_config(field, value):
|
|
with pytest.raises(LLMError): replace(provider(), **{field:value})
|
|
|
|
|
|
def test_config_defaults_and_loading(tmp_path):
|
|
config = GatewayConfig.load("config/llm.example.toml")
|
|
assert config.max_attempts == 10
|
|
assert [p.id for p in config.providers] == ["local","own-server","cloud"]
|
|
assert config.providers[0].enabled
|
|
assert not config.providers[-1].enabled
|
|
invalid = tmp_path / "invalid.toml"
|
|
invalid.write_text('[llm]\nmax_attempts=true\nproviders=[]')
|
|
with pytest.raises(LLMError): GatewayConfig.load(invalid)
|
|
with pytest.raises(LLMError): GatewayConfig((provider(),provider()))
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["external_schema","bad_schema","privacy","messages","limit"])
|
|
def test_request_validated_before_network(change):
|
|
adapter = Adapter({"first":success()})
|
|
req = request()
|
|
cfg = gateway(adapter)
|
|
if change == "external_schema": req["output_schema"] = {"type":"object","$ref":"https://example.org/schema"}
|
|
if change == "bad_schema": req["output_schema"] = {"type":"wrong"}
|
|
if change == "privacy": req["privacy"] = "unknown"
|
|
if change == "messages": req["messages"] = [{"role":"tool","content":"execute"}]
|
|
if change == "limit": cfg = gateway(adapter,max_request_bytes=1)
|
|
with pytest.raises(LLMError): asyncio.run(cfg.complete_round(req))
|
|
assert not adapter.calls
|
|
|
|
|
|
@pytest.fixture
|
|
def http_endpoint():
|
|
seen = []
|
|
reply = {"status":200,"body":{"model":"remote-model","choices":[{"finish_reason":"stop","message":{"content":'{"answer":"ok"}'}}],"usage":{"prompt_tokens":3,"completion_tokens":5}},"headers":{}}
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
seen.append({"path":self.path,"body":json.loads(self.rfile.read(int(self.headers["Content-Length"]))),"authorization":self.headers.get("Authorization")})
|
|
self.send_response(reply["status"])
|
|
for key,value in reply["headers"].items(): self.send_header(key,value)
|
|
self.end_headers()
|
|
body = reply["body"]
|
|
self.wfile.write(body if isinstance(body,bytes) else json.dumps(body).encode())
|
|
def log_message(self, *args): pass
|
|
server = ThreadingHTTPServer(("127.0.0.1",0),Handler)
|
|
thread = threading.Thread(target=server.serve_forever,daemon=True)
|
|
thread.start()
|
|
yield f"http://127.0.0.1:{server.server_port}/v1",seen,reply
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
|
|
@pytest.mark.parametrize("format", ["json_schema","json_object","none"])
|
|
def test_real_http_adapter(http_endpoint, monkeypatch, format):
|
|
url, seen, _ = http_endpoint
|
|
monkeypatch.setenv("TEST_API_KEY","test-secret")
|
|
config = GatewayConfig((ProviderConfig(id="test",model="chosen-model",base_url=url,api_key_env="TEST_API_KEY",response_format=format),))
|
|
output = asyncio.run(LLMGateway(config).complete_round(request()))
|
|
assert output["document"] == {"answer":"ok"}
|
|
assert output["metadata"]["model"] == "remote-model"
|
|
assert output["metadata"]["input_tokens"] == 3
|
|
assert seen[0]["path"] == "/v1/chat/completions"
|
|
assert seen[0]["authorization"] == "Bearer test-secret"
|
|
assert seen[0]["body"]["stream"] is False
|
|
assert seen[0]["body"]["model"] == "chosen-model"
|
|
assert "test-secret" not in json.dumps(output)
|
|
if format == "none": assert "response_format" not in seen[0]["body"]
|
|
else: assert seen[0]["body"]["response_format"]["type"] == format
|
|
|
|
|
|
@pytest.mark.parametrize("status,temporary", [(401,False),(400,False),(408,True),(429,True),(500,True),(503,True),(302,False)])
|
|
def test_http_errors_and_redirects(http_endpoint,status,temporary):
|
|
url, seen, reply = http_endpoint
|
|
reply.update(status=status,body=b"secret error body",headers={"Location":url+"/redirect"})
|
|
cfg = GatewayConfig((ProviderConfig(id="test",model="m",base_url=url),))
|
|
with pytest.raises(GatewayUnavailable if temporary else LLMError) as caught:
|
|
asyncio.run(LLMGateway(cfg).complete_round(request()))
|
|
assert "secret" not in str(caught.value)
|
|
assert len(seen) == 1
|
|
if temporary:
|
|
assert caught.value.attempts[0]["http_status"] == status
|
|
else:
|
|
assert caught.value.http_status == status
|
|
|
|
|
|
@pytest.mark.parametrize("status", [401, 429, 503])
|
|
def test_http_status_persisted_and_visible_in_safe_telemetry(make_package, tmp_path, status):
|
|
from viano_bug_analysis.telemetry import Telemetry
|
|
from viano_bug_analysis.telemetry.__main__ import read_events
|
|
from viano_bug_analysis.case.contracts import validate
|
|
workspace = Workspace(tmp_path / "work")
|
|
error = ProviderUnavailable("LLM_HTTP_TEMPORARY", http_status=status) if status != 401 else LLMError("LLM_HTTP_CONFIGURATION", http_status=status)
|
|
with Telemetry(workspace) as telemetry:
|
|
store = CheckpointStore(workspace, telemetry=telemetry)
|
|
store.intake(make_package())
|
|
config = GatewayConfig((provider(),))
|
|
gateway = LLMGateway(config, {"fixture": Adapter({"first": error})})
|
|
result = asyncio.run(CaseLLMRunner(store, gateway).run("case-A", "stable-package", operation_id="http-status", request=request()))
|
|
with PackageReader(result.package) as reader:
|
|
record = json.loads(reader.read_bytes(result.result_ref))
|
|
if status == 401:
|
|
assert record["outcome"]["http_status"] == status
|
|
del record["outcome"]["http_status"]
|
|
else:
|
|
assert record["outcome"]["attempts"][0]["http_status"] == status
|
|
del record["outcome"]["attempts"][0]["http_status"]
|
|
# Optional field: previously saved v1 records remain valid.
|
|
validate(record, "llm-call")
|
|
events = read_events(workspace.root)
|
|
event = next(e for e in events if e["event"] == "llm_round")
|
|
if status == 401:
|
|
assert event["detail"]["http_status"] == status
|
|
else:
|
|
assert event["detail"]["failed_providers"][0]["http_status"] == status
|
|
|
|
|
|
def test_successful_failover_keeps_first_http_status():
|
|
config = GatewayConfig((provider("first"), provider("second")))
|
|
adapter = Adapter({"first": ProviderUnavailable("LLM_HTTP_TEMPORARY", http_status=429),
|
|
"second": ProviderResponse({"answer": "ok"}, "fixture-model", 1, 1)})
|
|
outcome = asyncio.run(LLMGateway(config, {"fixture": adapter}).complete_round(request()))
|
|
assert outcome["metadata"]["attempts"] == [
|
|
{"provider": "first", "code": "LLM_HTTP_TEMPORARY", "http_status": 429}]
|
|
|
|
|
|
@pytest.mark.parametrize("value", [True, "secret-value", 99, 600, None])
|
|
def test_http_status_only_accepts_bounded_integers(value):
|
|
from viano_bug_analysis.telemetry.safety import detail
|
|
assert LLMError("LLM_HTTP_CONFIGURATION", http_status=value).http_status is None
|
|
data = detail("llm_round", {"http_status": value, "failed_providers": [{"http_status": value}]})
|
|
assert data["http_status"] is None
|
|
assert data["failed_providers"][0]["http_status"] is None
|
|
assert "secret-value" not in json.dumps(data)
|
|
|
|
|
|
@pytest.mark.parametrize("body", [b'not JSON',b'{"choices":[],"choices":[]}',
|
|
{"choices":[{"finish_reason":"length","message":{"content":"{}"}}]},
|
|
{"choices":[{"finish_reason":"stop","message":{"refusal":"no","content":"{}"}}]},
|
|
{"choices":[{"finish_reason":"stop","message":{"content":'{"answer":NaN}'}}]}])
|
|
def test_bad_http_responses(http_endpoint,body):
|
|
url, _, reply = http_endpoint
|
|
reply["body"] = body
|
|
cfg = GatewayConfig((ProviderConfig(id="test",model="m",base_url=url),))
|
|
with pytest.raises(LLMError): asyncio.run(LLMGateway(cfg).complete_round(request()))
|
|
|
|
|
|
def test_response_size_limit(http_endpoint):
|
|
url, _, reply = http_endpoint
|
|
reply["body"] = b"x" * 1000
|
|
cfg = GatewayConfig((ProviderConfig(id="test",model="m",base_url=url),), max_response_bytes=100)
|
|
with pytest.raises(LLMError,match="LLM_RESPONSE_LIMIT"): asyncio.run(LLMGateway(cfg).complete_round(request()))
|
|
|
|
|
|
@pytest.fixture
|
|
def active_case(make_package,tmp_path):
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
original = make_package(entries={"original/data":b"keep","unknown/vendor":b"opaque"})
|
|
store.intake(original)
|
|
active = store.advance("case-A","stable-package",State.BUILDING_CASE,step_id="start")
|
|
return store,active,original
|
|
|
|
|
|
def test_ten_rounds_persist_and_block_without_sleep(active_case,caplog):
|
|
store, active, original = active_case
|
|
initial = original.read_bytes()
|
|
adapter = Adapter({"first":ProviderUnavailable("LLM_CONNECTION"),"second":ProviderUnavailable("LLM_CONNECTION")})
|
|
cfg = gateway(adapter,retry_delay_seconds=1)
|
|
now = [datetime(2026,1,1,tzinfo=timezone.utc)]
|
|
for index in range(10):
|
|
runner = CaseLLMRunner(CheckpointStore(store.workspace),cfg,clock=lambda:now[0])
|
|
outcome = asyncio.run(runner.run("case-A","stable-package",operation_id="operation",request=request()))
|
|
assert outcome.rounds == index+1
|
|
assert outcome.status == ("BLOCKED" if index == 9 else "RETRY_PENDING")
|
|
deferred = asyncio.run(runner.run("case-A","stable-package",operation_id="operation",request=request()))
|
|
assert deferred.status in {"DEFERRED","BLOCKED"}
|
|
now[0] += timedelta(seconds=1)
|
|
assert adapter.calls == ["first","second"] * 10
|
|
with PackageReader(outcome.package) as reader:
|
|
assert reader.manifest["state"] == State.BLOCKED
|
|
assert reader.manifest["routing"]["reason"] == "LLM_UNAVAILABLE"
|
|
assert reader.manifest["routing"]["destination"] == "inbox"
|
|
assert reader.manifest["analysis_revision"] == 1
|
|
assert reader.read_bytes("original/data") == b"keep"
|
|
assert reader.read_bytes("unknown/vendor") == b"opaque"
|
|
assert len([p for p in reader.files if "/round-" in p]) == 10
|
|
assert original.read_bytes() == initial
|
|
assert PackageValidator().validate(outcome.package).valid
|
|
assert "Casebuilding failed: no reachable LLM server" in caplog.text
|
|
lines = (store.workspace.runtime / "logs/llm.jsonl").read_text().splitlines()
|
|
assert any("status=BLOCKED" in json.loads(line)["message"] for line in lines)
|
|
# Exhaustion stays blocked even after the cooldown; inbox cannot start an endless campaign.
|
|
assert not any(r.status == "RESUMED" for r in InboxProcessor(store.workspace).poll())
|
|
|
|
|
|
def test_success_replay_no_network_and_config_conflict(active_case):
|
|
store, _, _ = active_case
|
|
adapter = Adapter({"first":success()})
|
|
cfg = gateway(adapter)
|
|
runner = CaseLLMRunner(store,cfg)
|
|
first = asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
second = asyncio.run(CaseLLMRunner(CheckpointStore(store.workspace),cfg).run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert second == first
|
|
assert adapter.calls == ["first"]
|
|
changed = request()
|
|
changed["messages"][0]["content"] += "changed"
|
|
with pytest.raises(LLMError,match="LLM_OPERATION_CONFLICT"):
|
|
asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=changed))
|
|
|
|
|
|
def test_schema_error_is_persisted_not_blocked(active_case):
|
|
store, _, _ = active_case
|
|
adapter = Adapter({"first":ProviderResponse({"wrong":1},"m")})
|
|
runner = CaseLLMRunner(store,gateway(adapter))
|
|
failed = asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert failed.status == "ERROR" and failed.code == "LLM_OUTPUT_SCHEMA"
|
|
with PackageReader(failed.package) as reader: assert reader.manifest["state"] == State.BUILDING_CASE
|
|
assert asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request())) == failed
|
|
assert adapter.calls == ["first"]
|
|
|
|
|
|
def test_duplicate_dispatch_lock_does_not_wait(active_case):
|
|
store, _, _ = active_case
|
|
entered = asyncio.Event()
|
|
release = asyncio.Event()
|
|
class Waiting:
|
|
async def generate(self,*args,**kwargs):
|
|
entered.set()
|
|
await release.wait()
|
|
return success()
|
|
runner = CaseLLMRunner(store,LLMGateway(GatewayConfig((provider(),)),{"fixture":Waiting()}))
|
|
async def dispatch():
|
|
task = asyncio.create_task(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
await entered.wait()
|
|
with pytest.raises(LockBusy): await runner.run("case-A","stable-package",operation_id="op",request=request())
|
|
release.set()
|
|
return await task
|
|
assert asyncio.run(dispatch()).status == "SUCCESS"
|
|
|
|
|
|
def test_completed_casebuilder_ignores_gateway_artifacts(make_package,tmp_path):
|
|
store = CheckpointStore(Workspace(tmp_path / "store"))
|
|
store.intake(make_package(entries={"unknown":b"data"}))
|
|
builder = CaseBuilder(store)
|
|
built = builder.build("case-A","stable-package")
|
|
runner = CaseLLMRunner(store,gateway(Adapter({"first":success()})))
|
|
outcome = asyncio.run(runner.run("case-A","stable-package",operation_id="transport-test",request=request()))
|
|
assert outcome.status == "SUCCESS"
|
|
assert builder.build("case-A","stable-package").reused
|
|
|
|
|
|
def test_configured_cloud_and_narrower_request_policy():
|
|
adapter = Adapter({"cloud":success()})
|
|
config = GatewayConfig((replace(provider("cloud"), privacy="cloud", base_url="https://cloud.example/v1"),), privacy_profile="cloud_allowed")
|
|
req = PromptBuilder().build(task="test",data={},output_schema=SCHEMA)
|
|
assert asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(req))["metadata"]["provider"] == "cloud"
|
|
with pytest.raises(LLMError,match="LLM_PRIVACY_NO_PROVIDER"):
|
|
asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request()))
|
|
|
|
|
|
def test_disabled_provider_is_never_called():
|
|
adapter = Adapter({"second":success()})
|
|
config = GatewayConfig((replace(provider(),enabled=False),provider("second")))
|
|
asyncio.run(LLMGateway(config,{"fixture":adapter}).complete_round(request()))
|
|
assert adapter.calls == ["second"]
|
|
|
|
|
|
def test_other_case_progresses_while_endpoint_waits(make_package,manifest,tmp_path):
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
for identity in ("case-A","case-B"):
|
|
doc = dict(manifest,case_id=identity)
|
|
store.intake(make_package(document=doc,name=identity))
|
|
store.advance(identity,"stable-package",State.BUILDING_CASE,step_id="start")
|
|
async def concurrent():
|
|
entered, release = asyncio.Event(), asyncio.Event()
|
|
class Slow:
|
|
async def generate(self,*args,**kwargs):
|
|
entered.set()
|
|
await release.wait()
|
|
return success()
|
|
slow = CaseLLMRunner(store,LLMGateway(GatewayConfig((provider(),)),{"fixture":Slow()}))
|
|
fast = CaseLLMRunner(store,gateway(Adapter({"first":success()})))
|
|
pending = asyncio.create_task(slow.run("case-A","stable-package",operation_id="op",request=request()))
|
|
await entered.wait()
|
|
finished = await fast.run("case-B","stable-package",operation_id="op",request=request())
|
|
assert finished.status == "SUCCESS"
|
|
assert not pending.done()
|
|
release.set()
|
|
assert (await pending).status == "SUCCESS"
|
|
asyncio.run(concurrent())
|
|
|
|
|
|
def test_pending_retry_recovers_then_succeeds(active_case):
|
|
store, _, _ = active_case
|
|
adapter = Adapter({"first":ProviderUnavailable("LLM_CONNECTION"),"second":ProviderUnavailable("LLM_CONNECTION")})
|
|
cfg = gateway(adapter,retry_delay_seconds=2)
|
|
now = [datetime(2026,1,1,tzinfo=timezone.utc)]
|
|
runner = CaseLLMRunner(store,cfg,clock=lambda:now[0])
|
|
first = asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert first.status == "RETRY_PENDING"
|
|
now[0] += timedelta(seconds=2)
|
|
adapter.results["first"] = success()
|
|
resumed = CaseLLMRunner(CheckpointStore(store.workspace),cfg,clock=lambda:now[0])
|
|
outcome = asyncio.run(resumed.run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert outcome.status == "SUCCESS" and outcome.rounds == 2
|
|
assert adapter.calls == ["first","second","first"]
|
|
|
|
|
|
def test_exhaustion_flag_survives_returning_package_intake(active_case):
|
|
store, _, _ = active_case
|
|
adapter = Adapter({"first":ProviderUnavailable("LLM_CONNECTION"),"second":ProviderUnavailable("LLM_CONNECTION")})
|
|
runner = CaseLLMRunner(store,gateway(adapter,max_attempts=1),clock=lambda:datetime(2026,1,1,tzinfo=timezone.utc))
|
|
blocked = asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert blocked.status == "BLOCKED"
|
|
# A new service receives the self-contained blocked package after its due date.
|
|
receiving = CheckpointStore(Workspace(store.workspace.root / "new-service"))
|
|
receiving.intake(blocked.package)
|
|
assert not any(r.status == "RESUMED" for r in InboxProcessor(receiving.workspace).poll())
|
|
|
|
|
|
def test_missing_credentials_never_sent(http_endpoint,monkeypatch):
|
|
url, seen, _ = http_endpoint
|
|
monkeypatch.delenv("NONEXISTENT_LLM_KEY",raising=False)
|
|
cfg = GatewayConfig((ProviderConfig(id="test",model="m",base_url=url,api_key_env="NONEXISTENT_LLM_KEY"),))
|
|
with pytest.raises(LLMError,match="LLM_CREDENTIALS"):
|
|
asyncio.run(LLMGateway(cfg).complete_round(request()))
|
|
assert not seen
|
|
|
|
|
|
def test_crash_after_publication_reuses_and_repairs_metadata(active_case,monkeypatch):
|
|
store, _, _ = active_case
|
|
adapter = Adapter({"first":success()})
|
|
runner = CaseLLMRunner(store,gateway(adapter))
|
|
def crash(*args): raise RuntimeError("simulated publication/index crash")
|
|
monkeypatch.setattr(store,"_index",crash)
|
|
with pytest.raises(RuntimeError,match="simulated"):
|
|
asyncio.run(runner.run("case-A","stable-package",operation_id="op",request=request()))
|
|
recovered_store = CheckpointStore(store.workspace)
|
|
recovered = asyncio.run(CaseLLMRunner(recovered_store,gateway(adapter)).run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert recovered.status == "SUCCESS"
|
|
assert adapter.calls == ["first"]
|
|
assert json.loads((store.workspace.runtime / "current_job.json").read_bytes()) is None
|
|
assert json.loads((recovered.package.parent / "current.json").read_bytes())["package"] == recovered.package.name
|
|
|
|
|
|
@pytest.mark.parametrize("contract", ["llm-step", "llm-call"])
|
|
def test_llm_schemas_valid(contract):
|
|
from importlib.resources import files
|
|
from jsonschema import Draft202012Validator
|
|
Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath(contract + "-v1.schema.json").read_text()))
|
|
|
|
|
|
def test_unrelated_blocked_case_never_calls_llm(active_case):
|
|
store, _, _ = active_case
|
|
store.advance("case-A","stable-package",State.BLOCKED,step_id="research-offline",
|
|
reason="RESEARCH_UNAVAILABLE",retry_not_before="2099-01-01T00:00:00Z")
|
|
adapter = Adapter({"first":success()})
|
|
outcome = asyncio.run(CaseLLMRunner(store,gateway(adapter)).run("case-A","stable-package",operation_id="op",request=request()))
|
|
assert outcome.status == "BLOCKED" and outcome.code == "CASE_BLOCKED"
|
|
assert not adapter.calls
|