190 lines
8.3 KiB
Python
190 lines
8.3 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():
|
|
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_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("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")
|
|
assert suite_api.credential("Bearer local-secret", tmp_path)[1] == []
|
|
assert suite_api.credential("Bearer synthetic-secret", tmp_path)[1] == ["claude", "codex"]
|
|
with pytest.raises(Problem, match="authentication"):
|
|
suite_api.credential("Bearer invalid", tmp_path)
|