255 lines
12 KiB
Python
255 lines
12 KiB
Python
"""Critical policy, completeness, isolation, and retry guarantees for suite jobs."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "services/hermes/scripts"))
|
|
import suite_api
|
|
import suite_backends
|
|
from suite_contract import MODELS, Problem, preflight, prompt, validate_request, validate_result
|
|
from suite_jobs import Jobs
|
|
from suite_synthetic import fixture, score
|
|
|
|
|
|
def external(size=14):
|
|
request = fixture(size)[0]
|
|
request["routing"] = {"allow_external": True, "allowed_external_providers": ["claude"]}
|
|
return suite_api.authorize(request, ["claude"])
|
|
|
|
|
|
def expected_result(size):
|
|
_, expected = fixture(size)
|
|
groups = {}
|
|
for alias, family in expected.items():
|
|
groups.setdefault(family, []).append(alias)
|
|
return {"groups": [{"name": name, "description": "Shared synthetic implementation machinery",
|
|
"members": members} for name, members in groups.items()]}
|
|
|
|
|
|
@pytest.mark.parametrize("size", [14, 75, 363])
|
|
def test_full_suite_capacity_and_coverage(size):
|
|
request = external(size)
|
|
selection = preflight(request)
|
|
assert selection["provider"] == "claude"
|
|
assert selection["case_count"] == size
|
|
# Every field and record, including duplicate text, survives preparation exactly.
|
|
assert json.loads(prompt(request))["cases"] == request["cases"]
|
|
assert len({case["alias"] for case in request["cases"]}) == size
|
|
result = validate_result(expected_result(size), request)
|
|
assert score(result, fixture(size)[1])["pair_recall"] == 1
|
|
result["groups"][0]["members"].append("[reference]")
|
|
with pytest.raises(Problem, match="invalid_case_assignments"):
|
|
validate_result(result, request)
|
|
|
|
|
|
@pytest.mark.parametrize("policy", [
|
|
{"allow_external": "false"}, {"allow_external": 1},
|
|
{"allow_external": False, "allowed_external_providers": ["claude"]},
|
|
{"allow_external": True, "allowed_external_providers": []},
|
|
{"allow_external": True, "allowed_external_providers": ["unknown"]},
|
|
{"allow_external": True, "allowed_external_providers": ["claude", "claude"]},
|
|
{"allow_external": True, "allowed_external_providers": "claude"},
|
|
])
|
|
def test_malformed_policy_rejected(policy):
|
|
request = fixture(14)[0]
|
|
request["routing"] = policy
|
|
with pytest.raises(Problem):
|
|
suite_api.authorize(request, ["claude"])
|
|
|
|
|
|
def test_permissions_and_external_data_scope(monkeypatch):
|
|
monkeypatch.delenv("PLANNING_GENERALIZED_CLAUDE_APPROVED", raising=False)
|
|
with pytest.raises(Problem, match="provider_forbidden"):
|
|
suite_api.authorize(external(), [])
|
|
request = external()
|
|
request["cases"][0]["description"] = "Changed input requires separate approval"
|
|
with pytest.raises(Problem, match="external_data_not_approved"):
|
|
suite_api.authorize(request, ["claude"])
|
|
|
|
|
|
def test_generalized_approval_remains_claude_and_credential_scoped(monkeypatch):
|
|
monkeypatch.setenv("PLANNING_GENERALIZED_CLAUDE_APPROVED", "true")
|
|
request = external()
|
|
request["cases"][0]["description"] = "Synthetic example of a generalized input"
|
|
with pytest.raises(Problem, match="external_data_not_approved"):
|
|
suite_api.authorize(request, ["claude"])
|
|
assert suite_api.authorize(request, ["claude"], operational=True)["cases"] == request["cases"]
|
|
with pytest.raises(Problem, match="provider_forbidden"):
|
|
suite_api.authorize(request, [], operational=True)
|
|
for providers in (["codex"], ["claude", "codex"]):
|
|
request["routing"]["allowed_external_providers"] = providers
|
|
with pytest.raises(Problem, match="provider_forbidden"):
|
|
suite_api.authorize(request, providers, operational=True)
|
|
|
|
|
|
def test_local_schema_excludes_descriptions_from_members(monkeypatch):
|
|
from types import SimpleNamespace
|
|
captured = []
|
|
request = validate_request({"campaign": "SYNTHETIC", "suite": "PARSER", "cases": [
|
|
{"alias": "CASE-1", "description": "Parse valid configuration text"}]}, [])
|
|
def respond(url, value, *args):
|
|
captured.append(value)
|
|
return {"done": True, "model": MODELS["local"]["model"], "response": '{"groups":[]}'}
|
|
monkeypatch.setattr(suite_backends, "post", respond)
|
|
monkeypatch.setattr(suite_backends, "Path", lambda _: SimpleNamespace(read_text=lambda: "fake"))
|
|
suite_backends.local_generate(request, threading.Event(), "192.168.22.8")
|
|
items = captured[0]["format"]["properties"]["groups"]["items"]["properties"]["members"]["items"]
|
|
assert items["enum"] == ["CASE-1"]
|
|
assert "enum" not in suite_backends.SCHEMA["properties"]["groups"]["items"]["properties"]["members"]["items"]
|
|
|
|
|
|
def test_local_default_cannot_overflow_to_provider():
|
|
request = validate_request(fixture(14)[0], ["claude", "codex"])
|
|
assert request["routing"] == {"allow_external": False, "allowed_external_providers": []}
|
|
with pytest.raises(Problem, match="capacity_or_unsupported_backend") as error:
|
|
preflight(request)
|
|
assert set(error.value.details["candidates"]) == {"local"}
|
|
|
|
|
|
def test_codex_unverified_capacity_fails_closed():
|
|
request = fixture(14)[0]
|
|
request["routing"] = {"allow_external": True, "allowed_external_providers": ["codex"]}
|
|
with pytest.raises(Problem, match="capacity_or_unsupported_backend"):
|
|
preflight(suite_api.authorize(request, ["codex"]))
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["duplicate", "ownership", "alias", "unknown", "strategy"])
|
|
def test_bad_membership_and_contract(mutation):
|
|
request = fixture(14)[0]
|
|
if mutation == "duplicate":
|
|
request["cases"][1]["alias"] = request["cases"][0]["alias"]
|
|
elif mutation == "ownership":
|
|
request["cases"][1]["suite"] = "different"
|
|
elif mutation == "alias":
|
|
request["cases"][0]["alias"] = "[reference]"
|
|
elif mutation == "unknown":
|
|
request["cases"][0]["raw"] = {"secret": "not permitted"}
|
|
else:
|
|
request["execution"] = {"strategy": "independent_batches"}
|
|
with pytest.raises(Problem):
|
|
validate_request(request, [])
|
|
|
|
|
|
def test_output_assignment_errors():
|
|
request = external()
|
|
for mutation in ("omitted", "duplicate", "invented"):
|
|
result = expected_result(14)
|
|
members = result["groups"][0]["members"]
|
|
if mutation == "omitted":
|
|
members.pop()
|
|
elif mutation == "duplicate":
|
|
members.append(members[0])
|
|
else:
|
|
members[0] = "CASE-INVENTED"
|
|
with pytest.raises(Problem):
|
|
validate_result(result, request)
|
|
|
|
|
|
def test_idempotency_ownership_busy_restart(tmp_path):
|
|
jobs = Jobs(tmp_path / "jobs.sqlite")
|
|
request = external()
|
|
selection = preflight(request)
|
|
first, created = jobs.submit("owner", "same-key", request, selection, "192.168.22.8", launch=False)
|
|
assert created
|
|
repeat, created = jobs.submit("owner", "same-key", request, selection, "192.168.22.8", launch=False)
|
|
assert not created and repeat["job_id"] == first["job_id"]
|
|
with pytest.raises(Problem, match="idempotency_conflict"):
|
|
jobs.submit("owner", "same-key", external(75), selection, "192.168.22.8", launch=False)
|
|
with pytest.raises(Problem, match="capacity_busy"):
|
|
jobs.submit("owner", "another-key", request, selection, "192.168.22.8", launch=False)
|
|
with pytest.raises(Problem, match="job_not_found"):
|
|
jobs.get(first["job_id"], "other-owner")
|
|
restarted = Jobs(tmp_path / "jobs.sqlite")
|
|
assert restarted.get(first["job_id"], "owner")["error"]["code"] == "interrupted_no_retry"
|
|
assert not restarted.submit("owner", "same-key", request, selection, "192.168.22.8", launch=False)[1]
|
|
assert b"thermal chamber" not in (tmp_path / "jobs.sqlite").read_bytes()
|
|
|
|
|
|
def test_no_fallback_after_local_failure(tmp_path, monkeypatch, capsys):
|
|
request = {"campaign": "SYNTHETIC", "suite": "TINY", "cases": [
|
|
{"alias": "CASE-1", "description": "Read parser status"}]}
|
|
request = validate_request(request, [])
|
|
selected = preflight(request)
|
|
calls = []
|
|
monkeypatch.setattr(suite_backends, "switchyard_decision", lambda provider: calls.append(provider))
|
|
def fail(*args):
|
|
raise Problem("backend_unavailable", 503)
|
|
monkeypatch.setattr(suite_backends, "local_generate", fail)
|
|
monkeypatch.setattr(suite_backends, "claude_generate", lambda *args: pytest.fail("external launch"))
|
|
jobs = Jobs(tmp_path / "jobs.sqlite")
|
|
document, _ = jobs.submit("owner", "local-failure", request, selected, "192.168.22.8", launch=False)
|
|
jobs.run(document["job_id"], "owner", request, selected, "192.168.22.8")
|
|
assert calls == ["local"]
|
|
assert jobs.get(document["job_id"], "owner")["status"] == "failed"
|
|
assert "Read parser status" not in capsys.readouterr().out
|
|
|
|
|
|
def test_fresh_cli_isolation_and_schema():
|
|
command = suite_backends.claude_command(MODELS["claude"]["model"], 5)
|
|
assert "--safe-mode" in command and "--no-session-persistence" in command
|
|
assert not any("bypass" in flag or "resume" in flag for flag in command)
|
|
assert command[command.index("--tools") + 1] == ""
|
|
env = suite_backends.claude_environment("/jobs/fresh", "fake-token")
|
|
assert env["DISABLE_COMPACT"] == "1"
|
|
assert env["CLAUDE_CODE_MAX_RETRIES"] == "0"
|
|
assert "ANTHROPIC_API_KEY" not in env
|
|
|
|
|
|
def test_compaction_and_incomplete_detection():
|
|
for events, code in [([{"type": "system", "subtype": "compact_boundary"}], "compaction_detected"),
|
|
([], "incomplete_generation")]:
|
|
with pytest.raises(Problem, match=code):
|
|
suite_backends.parse_claude("\n".join(json.dumps(e) for e in events), "claude-fable-5")
|
|
|
|
|
|
@pytest.mark.parametrize("failure,code", [
|
|
(None, None), ("model", "model_changed"),
|
|
("context", "backend_capabilities_changed"),
|
|
("tools", "worker_isolation_failed"), ("output", "incomplete_generation"),
|
|
])
|
|
def test_actual_cli_envelope_and_runtime_guards(failure, code):
|
|
model = MODELS["claude"]["model"]
|
|
init = {"type": "system", "subtype": "init", "model": model + "[1m]",
|
|
"tools": ["StructuredOutput"], "mcp_servers": [], "plugins": []}
|
|
limits = {"contextWindow": 1000000, "maxOutputTokens": 64000,
|
|
"canonicalModel": model, "provider": "firstParty"}
|
|
result = {"type": "result", "subtype": "success", "is_error": False,
|
|
"modelUsage": {model + "[1m]": limits}, "usage": {},
|
|
"structured_output": expected_result(14)}
|
|
if failure == "model":
|
|
result["modelUsage"] = {"claude-unexpected": limits}
|
|
elif failure == "context":
|
|
limits["contextWindow"] = 200000
|
|
elif failure == "tools":
|
|
init["tools"].append("Bash")
|
|
elif failure == "output":
|
|
result["stop_reason"] = "max_tokens"
|
|
raw = "\n".join(json.dumps(e) for e in [init, result])
|
|
if code:
|
|
with pytest.raises(Problem, match=code):
|
|
suite_backends.parse_claude(raw, model)
|
|
else:
|
|
value, metadata = suite_backends.parse_claude(raw, model)
|
|
assert value == expected_result(14) and metadata["model"] == model
|
|
|
|
|
|
@pytest.mark.parametrize("raw", ['{"routing":{},"routing":{}}', '{"x":NaN}', '{"x":Infinity}', '{'])
|
|
def test_strict_json(raw):
|
|
with pytest.raises(Problem, match="invalid_json"):
|
|
suite_api.strict_json(raw)
|
|
|
|
|
|
def test_credential_permissions(tmp_path):
|
|
(tmp_path / "token").write_text("local-secret")
|
|
(tmp_path / "synthetic-token").write_text("synthetic-secret")
|
|
(tmp_path / "operational-token").write_text("operational-secret")
|
|
assert suite_api.credential("Bearer local-secret", tmp_path)[1] == []
|
|
assert suite_api.credential("Bearer synthetic-secret", tmp_path)[1] == ["claude", "codex"]
|
|
assert suite_api.credential("Bearer operational-secret", tmp_path)[1] == ["claude"]
|
|
with pytest.raises(Problem, match="authentication"):
|
|
suite_api.credential("Bearer invalid", tmp_path)
|