toki
de4d8f4ff8
세 Agent의 direct route를 동일한 fail-closed preflight와 격리 실행 경계에서 비교하고, 관측되지 않은 preset 셀이 실행되는 것을 막기 위해 연결 계약과 증거 수집 흐름을 고정한다.
600 lines
29 KiB
Python
600 lines
29 KiB
Python
"""Deterministic tests for the closed benchmark connectivity contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import FrozenInstanceError
|
|
from pathlib import Path
|
|
|
|
from scripts.agent_benchmark.connectivity import (
|
|
ISSUE_RESUME_CODES,
|
|
MAX_EVIDENCE_BYTES,
|
|
CallerCapability,
|
|
ConnectivityEvidenceError,
|
|
ConnectivityIssue,
|
|
ConnectivityResult,
|
|
ConnectivityValidationError,
|
|
EffectiveBinding,
|
|
RequestedEffectiveBinding,
|
|
canonical_evidence_bytes,
|
|
classify_issues,
|
|
make_result,
|
|
read_evidence,
|
|
validate_binding,
|
|
validate_result,
|
|
write_evidence,
|
|
)
|
|
from scripts.agent_benchmark.manifest import ExpectedBinding, IopCell, MatrixCell
|
|
|
|
EFFECTIVE_SCALARS = (
|
|
"effective_route_kind",
|
|
"effective_route_id",
|
|
"effective_model",
|
|
"effective_effort",
|
|
)
|
|
|
|
|
|
def _identity(label: str) -> str:
|
|
return "sha256:" + hashlib.sha256(label.encode("ascii")).hexdigest()
|
|
|
|
|
|
def _cell(route_kind: str = "direct") -> MatrixCell:
|
|
if route_kind == "direct":
|
|
bindings = (ExpectedBinding("request", "gpt", "xhigh"),)
|
|
else:
|
|
bindings = (
|
|
ExpectedBinding("selector", "gpt", "xhigh"),
|
|
ExpectedBinding("plan", "gpt", "high"),
|
|
ExpectedBinding("work", "gpt", None),
|
|
ExpectedBinding("review", "gpt", "high"),
|
|
)
|
|
return MatrixCell("cell", "codex", IopCell("gpt", "xhigh", route_kind, "route", bindings))
|
|
|
|
|
|
def _capability() -> CallerCapability:
|
|
return CallerCapability("codex", ("direct", "execution_preset"), ("high", "xhigh"))
|
|
|
|
|
|
def _requested_only(cell: MatrixCell) -> RequestedEffectiveBinding:
|
|
"""Blocked observation: requested identity only, no effective group at all."""
|
|
return RequestedEffectiveBinding(
|
|
cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
|
|
cell.iop.request_model, cell.iop.requested_effort,
|
|
)
|
|
|
|
|
|
def _binding(cell: MatrixCell) -> RequestedEffectiveBinding:
|
|
return RequestedEffectiveBinding(
|
|
cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
|
|
cell.iop.request_model, cell.iop.requested_effort,
|
|
cell.iop.route_kind, cell.iop.route_id,
|
|
cell.iop.request_model, cell.iop.requested_effort,
|
|
tuple(EffectiveBinding(item.stage, item.model, item.effort) for item in cell.iop.expected_bindings),
|
|
)
|
|
|
|
|
|
def _issue(code: str) -> ConnectivityIssue:
|
|
return ConnectivityIssue(code, ISSUE_RESUME_CODES[code])
|
|
|
|
|
|
class ConnectivityContractTest(unittest.TestCase):
|
|
def test_direct_and_preset_exact_contracts_are_frozen(self):
|
|
for route_kind in ("direct", "execution_preset"):
|
|
cell = _cell(route_kind)
|
|
result = make_result(cell, _capability(), _binding(cell))
|
|
self.assertEqual(result.status, "ready")
|
|
self.assertEqual(result.binding.effective_bindings, tuple(
|
|
EffectiveBinding(item.stage, item.model, item.effort)
|
|
for item in cell.iop.expected_bindings
|
|
))
|
|
with self.assertRaises(FrozenInstanceError):
|
|
result.status = "implementation_gap" # type: ignore[misc]
|
|
|
|
def test_every_requested_or_effective_substitution_fails_closed(self):
|
|
cell = _cell()
|
|
binding = _binding(cell)
|
|
substitutions = (
|
|
{"cell_id": "other"}, {"caller": "agy"},
|
|
{"requested_route_kind": "execution_preset"}, {"requested_route_id": "other"},
|
|
{"requested_model": "alias"}, {"requested_effort": "high"},
|
|
{"effective_route_kind": "execution_preset"}, {"effective_route_id": "other"},
|
|
{"effective_model": "alias"}, {"effective_effort": "high"},
|
|
)
|
|
for replacement in substitutions:
|
|
with self.subTest(replacement=replacement):
|
|
mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement})
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
validate_binding(cell, _capability(), mutated)
|
|
|
|
def test_missing_extra_and_reordered_stage_bindings_fail_closed(self):
|
|
cell = _cell("execution_preset")
|
|
binding = _binding(cell)
|
|
cases = (
|
|
binding.effective_bindings[:-1],
|
|
binding.effective_bindings + (EffectiveBinding("repair", "gpt", "high"),),
|
|
tuple(reversed(binding.effective_bindings)),
|
|
binding.effective_bindings[:-1] + (EffectiveBinding("review", "alias", "high"),),
|
|
)
|
|
for stages in cases:
|
|
with self.subTest(stages=stages):
|
|
observed = RequestedEffectiveBinding(**{**binding.__dict__, "effective_bindings": stages})
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
validate_binding(cell, _capability(), observed)
|
|
|
|
def test_capability_is_closed_and_requires_requested_effort(self):
|
|
cell = _cell()
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
validate_binding(cell, CallerCapability("codex", ("direct",), ("high",)), _binding(cell))
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
validate_binding(cell, CallerCapability("codex", ("execution_preset", "direct"), ("high", "xhigh")), _binding(cell))
|
|
|
|
def test_malformed_capability_entries_raise_closed_error(self):
|
|
cell = _cell()
|
|
binding = _binding(cell)
|
|
capabilities = (
|
|
CallerCapability("codex", ("unknown", "direct"), ("high", "xhigh")),
|
|
CallerCapability("codex", ("direct", ["execution_preset"]), ("high", "xhigh")),
|
|
CallerCapability("codex", ("direct", "direct"), ("high", "xhigh")),
|
|
CallerCapability("codex", (), ("high", "xhigh")),
|
|
CallerCapability("codex", ["direct"], ("high", "xhigh")),
|
|
CallerCapability("codex", ("direct",), ("high", 3)),
|
|
CallerCapability("codex", ("direct",), (["xhigh"],)),
|
|
CallerCapability("codex", ("direct",), ("xhigh", "high")),
|
|
CallerCapability("codex", ("direct",), ("Xhigh",)),
|
|
CallerCapability("unknown", ("direct",), ("xhigh",)),
|
|
"codex",
|
|
)
|
|
for capability in capabilities:
|
|
with self.subTest(capability=repr(capability)):
|
|
with self.assertRaises(ConnectivityValidationError) as caught:
|
|
validate_binding(cell, capability, binding)
|
|
self.assertNotIsInstance(caught.exception, (ValueError, TypeError, KeyError))
|
|
|
|
def test_classifier_is_closed_and_implementation_gap_has_precedence(self):
|
|
registration = (_issue("credential_missing"),)
|
|
implementation = (_issue("stream_incompatible"),)
|
|
self.assertEqual(classify_issues(()), "ready")
|
|
self.assertEqual(classify_issues(registration), "registration_required")
|
|
self.assertEqual(classify_issues(implementation), "implementation_gap")
|
|
self.assertEqual(classify_issues(registration + implementation), "implementation_gap")
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues((ConnectivityIssue("unknown", "register_credential"),))
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues([_issue("credential_missing")])
|
|
|
|
def test_issue_resume_pairs_are_closed_and_canonically_ordered(self):
|
|
for code, resume_code in ISSUE_RESUME_CODES.items():
|
|
with self.subTest(code=code):
|
|
self.assertIn(
|
|
classify_issues((ConnectivityIssue(code, resume_code),)),
|
|
("registration_required", "implementation_gap"),
|
|
)
|
|
for other_code, other_resume in ISSUE_RESUME_CODES.items():
|
|
if other_code == code:
|
|
continue
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues((ConnectivityIssue(code, other_resume),))
|
|
for text in ("register credential", "sk-live-0000", "", "register_credential ", None):
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues((ConnectivityIssue(code, text),))
|
|
canonical = (_issue("model_missing"), _issue("stream_incompatible"))
|
|
self.assertEqual(classify_issues(canonical), "implementation_gap")
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues(tuple(reversed(canonical)))
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
classify_issues(canonical + canonical[:1])
|
|
|
|
def test_malformed_issue_entries_raise_closed_error(self):
|
|
cell = _cell()
|
|
capability = _capability()
|
|
binding = _binding(cell)
|
|
valid_resume = ISSUE_RESUME_CODES["credential_missing"]
|
|
other_resume = ISSUE_RESUME_CODES["model_missing"]
|
|
bad_code_entries = (
|
|
ConnectivityIssue(["credential_missing"], valid_resume),
|
|
ConnectivityIssue({"code": "credential_missing"}, valid_resume),
|
|
ConnectivityIssue(42, valid_resume),
|
|
ConnectivityIssue(None, valid_resume),
|
|
ConnectivityIssue("unknown", valid_resume),
|
|
)
|
|
bad_resume_entries = (
|
|
ConnectivityIssue("credential_missing", ["register_credential"]),
|
|
ConnectivityIssue("credential_missing", {"resume_code": "register_credential"}),
|
|
ConnectivityIssue("credential_missing", 42),
|
|
ConnectivityIssue("credential_missing", None),
|
|
ConnectivityIssue("credential_missing", "register credential"),
|
|
ConnectivityIssue("credential_missing", "sk-live-0000"),
|
|
ConnectivityIssue("credential_missing", ""),
|
|
ConnectivityIssue("credential_missing", other_resume),
|
|
)
|
|
non_issue_objects = (
|
|
{"code": "credential_missing", "resume_code": valid_resume},
|
|
"credential_missing",
|
|
42,
|
|
None,
|
|
("credential_missing", valid_resume),
|
|
)
|
|
expected_message = {
|
|
"code": "invalid issue code",
|
|
"resume": "invalid issue resume_code",
|
|
"object": "invalid issue",
|
|
}
|
|
forbidden_raw_values = (
|
|
"sk-live-0000", "register credential", "credential_missing",
|
|
"register_credential", "register_model", "42", "None",
|
|
)
|
|
labelled = (
|
|
*(("code", item) for item in bad_code_entries),
|
|
*(("resume", item) for item in bad_resume_entries),
|
|
*(("object", item) for item in non_issue_objects),
|
|
)
|
|
for label, entry in labelled:
|
|
with self.subTest(label=label, entry=repr(entry)):
|
|
issues = (entry,)
|
|
with self.assertRaises(ConnectivityValidationError) as classifier_caught:
|
|
classify_issues(issues)
|
|
self.assertNotIsInstance(
|
|
classifier_caught.exception,
|
|
(TypeError, KeyError, ValueError, AttributeError),
|
|
)
|
|
self.assertEqual(str(classifier_caught.exception), expected_message[label])
|
|
for token in forbidden_raw_values:
|
|
self.assertNotIn(token, str(classifier_caught.exception))
|
|
self.assertNotIn(repr(entry), str(classifier_caught.exception))
|
|
with self.assertRaises(ConnectivityValidationError) as result_caught:
|
|
make_result(cell, capability, binding, issues)
|
|
self.assertNotIsInstance(
|
|
result_caught.exception,
|
|
(TypeError, KeyError, ValueError, AttributeError),
|
|
)
|
|
self.assertEqual(str(result_caught.exception), expected_message[label])
|
|
for token in forbidden_raw_values:
|
|
self.assertNotIn(token, str(result_caught.exception))
|
|
self.assertNotIn(repr(entry), str(result_caught.exception))
|
|
|
|
def test_ready_requires_complete_exact_effective_observation(self):
|
|
cell = _cell("execution_preset")
|
|
binding = _binding(cell)
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
make_result(cell, _capability(), _requested_only(cell))
|
|
partials = [{name: None} for name in EFFECTIVE_SCALARS]
|
|
partials.append({"effective_bindings": ()})
|
|
partials.append({name: None for name in EFFECTIVE_SCALARS})
|
|
partials.append({"effective_model": "alias"})
|
|
partials.append({"effective_bindings": binding.effective_bindings[:-1]})
|
|
for replacement in partials:
|
|
with self.subTest(replacement=replacement):
|
|
mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement})
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
make_result(cell, _capability(), mutated)
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
make_result(cell, _capability(), mutated, (_issue("model_missing"),))
|
|
|
|
def test_blocked_results_stay_blocked_and_cannot_be_forged_ready(self):
|
|
cell = _cell()
|
|
issues = (_issue("credential_missing"),)
|
|
blocked = make_result(cell, _capability(), _requested_only(cell), issues)
|
|
self.assertEqual(blocked.status, "registration_required")
|
|
self.assertIsNone(blocked.binding.effective_model)
|
|
self.assertEqual(blocked.binding.effective_bindings, ())
|
|
validate_result(cell, blocked)
|
|
forgeries = (
|
|
ConnectivityResult(_capability(), _requested_only(cell), issues, "ready"),
|
|
ConnectivityResult(_capability(), _binding(cell), issues, "ready"),
|
|
ConnectivityResult(_capability(), _binding(cell), (), "blocked"),
|
|
ConnectivityResult(_capability(), _requested_only(cell), (), "ready"),
|
|
)
|
|
for forged in forgeries:
|
|
with self.subTest(status=forged.status, issues=len(forged.issues)):
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
validate_result(cell, forged)
|
|
|
|
def test_no_contract_field_accepts_opaque_caller_text(self):
|
|
self.assertEqual(
|
|
{field.name for field in dataclasses.fields(ConnectivityIssue)},
|
|
{"code", "resume_code"},
|
|
)
|
|
cell = _cell()
|
|
binding = _binding(cell)
|
|
opaque = "sk-live-0000000000000000"
|
|
secret = "Authorization: Bearer should-not-appear"
|
|
for text in (secret, opaque, "https://private.invalid"):
|
|
with self.subTest(text=text):
|
|
with self.assertRaises(ConnectivityValidationError) as caught:
|
|
make_result(cell, _capability(), binding, (ConnectivityIssue("credential_missing", text),))
|
|
self.assertNotIn(text, str(caught.exception))
|
|
for replacement in (
|
|
{"requested_model": opaque}, {"effective_model": opaque},
|
|
{"effective_route_id": "private-endpoint"},
|
|
):
|
|
with self.subTest(replacement=replacement):
|
|
mutated = RequestedEffectiveBinding(**{**binding.__dict__, **replacement})
|
|
with self.assertRaises(ConnectivityValidationError) as caught:
|
|
make_result(cell, _capability(), mutated)
|
|
self.assertNotIn(opaque, str(caught.exception))
|
|
self.assertNotIn("private-endpoint", str(caught.exception))
|
|
|
|
|
|
class ConnectivityEvidenceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(dir="/tmp", prefix="connectivity-")
|
|
self.base = Path(os.path.realpath(self.temp.name))
|
|
self.root = self.base / "evidence"
|
|
self.root.mkdir()
|
|
self.cell = _cell()
|
|
self.result = make_result(self.cell, _capability(), _binding(self.cell))
|
|
self.endpoint_identity = _identity("endpoint")
|
|
self.config_identity = _identity("config")
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def _write(self, root, relative_path, cell=None, result=None):
|
|
write_evidence(
|
|
root, relative_path, cell or self.cell, result or self.result,
|
|
self.endpoint_identity, self.config_identity,
|
|
)
|
|
|
|
def test_canonical_evidence_is_deterministic_and_secret_safe(self):
|
|
one = canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, self.config_identity)
|
|
two = canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, self.config_identity)
|
|
self.assertEqual(one, two)
|
|
self.assertIn(self.endpoint_identity.encode(), one)
|
|
for forbidden in (b"Authorization", b"Bearer", b"https://", b"prompt", b"tool"):
|
|
self.assertNotIn(forbidden, one)
|
|
|
|
def test_blocked_results_omit_effective_observations_and_round_trip(self):
|
|
cases = (
|
|
("direct", _issue("model_missing"), "registration_required"),
|
|
("execution_preset", _issue("endpoint_incompatible"), "implementation_gap"),
|
|
)
|
|
for index, (route_kind, issue, status) in enumerate(cases):
|
|
with self.subTest(route_kind=route_kind):
|
|
cell = _cell(route_kind)
|
|
result = make_result(cell, _capability(), _requested_only(cell), (issue,))
|
|
self.assertEqual(result.status, status)
|
|
raw = canonical_evidence_bytes(cell, result, self.endpoint_identity, self.config_identity)
|
|
for absent in EFFECTIVE_SCALARS:
|
|
self.assertIn(f'"{absent}":null'.encode("ascii"), raw)
|
|
self.assertIn(b'"effective_bindings":[]', raw)
|
|
name = f"blocked-{index}.json"
|
|
self._write(self.root, name, cell, result)
|
|
parsed = read_evidence(self.root, name, cell)
|
|
self.assertEqual(parsed["status"], status)
|
|
self.assertIsNone(parsed["binding"]["effective_model"])
|
|
self.assertEqual(parsed["binding"]["effective_bindings"], [])
|
|
self.assertEqual(
|
|
parsed["issues"], [{"code": issue.code, "resume_code": issue.resume_code}]
|
|
)
|
|
forged = (
|
|
(self.root / name).read_text(encoding="ascii")
|
|
.replace(f'"status":"{status}"', '"status":"ready"')
|
|
.replace(f'{{"code":"{issue.code}","resume_code":"{issue.resume_code}"}}', "")
|
|
)
|
|
(self.root / f"forged-{index}.json").write_text(forged, encoding="ascii")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, f"forged-{index}.json", cell)
|
|
|
|
def test_write_is_no_overwrite_and_read_rejects_corruption(self):
|
|
self._write(self.root, "nested/preflight.json")
|
|
raw = (self.root / "nested/preflight.json").read_bytes()
|
|
self.assertEqual(read_evidence(self.root, "nested/preflight.json", self.cell)["status"], "ready")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, "nested/preflight.json")
|
|
(self.root / "nested/preflight.json").write_bytes(raw + b" ")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "nested/preflight.json", self.cell)
|
|
|
|
def test_reader_rejects_canonical_schema_drift(self):
|
|
self._write(self.root, "preflight.json")
|
|
raw = (self.root / "preflight.json").read_text(encoding="ascii")
|
|
(self.root / "preflight.json").write_text(raw.replace('"status":"ready"', '"status":"registration_required"'), encoding="ascii")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "preflight.json", self.cell)
|
|
|
|
def test_reader_rejects_canonical_binding_semantic_substitution(self):
|
|
def _reseat(parsed: dict) -> bytes:
|
|
return (
|
|
json.dumps(
|
|
parsed, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("ascii")
|
|
+ b"\n"
|
|
)
|
|
|
|
def _write_mutant(cell, result, name, mutate):
|
|
self._write(self.root, f"{name}.json", cell, result)
|
|
baseline = read_evidence(self.root, f"{name}.json", cell)
|
|
self.assertEqual(baseline["status"], result.status)
|
|
original = (self.root / f"{name}.json").read_text(encoding="ascii")
|
|
parsed = json.loads(original)
|
|
mutant = copy.deepcopy(parsed)
|
|
mutate(mutant)
|
|
mutant_name = f"mutant-{name}.json"
|
|
(self.root / mutant_name).write_bytes(_reseat(mutant))
|
|
return mutant_name
|
|
|
|
cases: list[tuple] = []
|
|
|
|
direct_cell = _cell("direct")
|
|
direct_result = make_result(direct_cell, _capability(), _binding(direct_cell))
|
|
direct_scalars = (
|
|
("requested_route_kind", "execution_preset"),
|
|
("requested_route_id", "route2"),
|
|
("requested_model", "gpth"),
|
|
("requested_effort", "high"),
|
|
("effective_route_kind", "execution_preset"),
|
|
("effective_route_id", "route2"),
|
|
("effective_model", "gpth"),
|
|
("effective_effort", "high"),
|
|
)
|
|
for field, value in direct_scalars:
|
|
cases.append((
|
|
"direct-ready-scalar", direct_cell, direct_result, f"direct-{field}",
|
|
lambda p, f=field, v=value: p["binding"].__setitem__(f, v),
|
|
))
|
|
cases.append((
|
|
"direct-ready-stage", direct_cell, direct_result, "direct-stage",
|
|
lambda p: p["binding"]["effective_bindings"].__setitem__(
|
|
0, {"stage": "request", "model": "gpth", "effort": "xhigh"}
|
|
),
|
|
))
|
|
|
|
preset_cell = _cell("execution_preset")
|
|
preset_result = make_result(preset_cell, _capability(), _binding(preset_cell))
|
|
preset_scalars = (
|
|
("requested_route_kind", "direct"),
|
|
("requested_route_id", "route2"),
|
|
("requested_model", "gpth"),
|
|
("requested_effort", "high"),
|
|
("effective_route_kind", "direct"),
|
|
("effective_route_id", "route2"),
|
|
("effective_model", "gpth"),
|
|
("effective_effort", "high"),
|
|
)
|
|
for field, value in preset_scalars:
|
|
cases.append((
|
|
"preset-ready-scalar", preset_cell, preset_result, f"preset-{field}",
|
|
lambda p, f=field, v=value: p["binding"].__setitem__(f, v),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-model", preset_cell, preset_result, "preset-stage-model",
|
|
lambda p: p["binding"]["effective_bindings"].__setitem__(
|
|
0, {"stage": "selector", "model": "gpth", "effort": "xhigh"}
|
|
),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-effort", preset_cell, preset_result, "preset-stage-effort",
|
|
lambda p: p["binding"]["effective_bindings"].__setitem__(
|
|
1, {"stage": "plan", "model": "gpt", "effort": "xhigh"}
|
|
),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-set", preset_cell, preset_result, "preset-stage-set",
|
|
lambda p: p["binding"]["effective_bindings"].__setitem__(
|
|
0, {"stage": "request", "model": "gpt", "effort": "xhigh"}
|
|
),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-order", preset_cell, preset_result, "preset-stage-order",
|
|
lambda p: p["binding"]["effective_bindings"].reverse(),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-extra", preset_cell, preset_result, "preset-stage-extra",
|
|
lambda p: p["binding"]["effective_bindings"].append(
|
|
{"stage": "repair", "model": "gpt", "effort": "high"}
|
|
),
|
|
))
|
|
cases.append((
|
|
"preset-ready-stage-missing", preset_cell, preset_result, "preset-stage-missing",
|
|
lambda p: p["binding"]["effective_bindings"].pop(),
|
|
))
|
|
|
|
blocked_cell = _cell("execution_preset")
|
|
blocked_result = make_result(
|
|
blocked_cell, _capability(), _binding(blocked_cell), (_issue("credential_missing"),)
|
|
)
|
|
self.assertEqual(blocked_result.status, "registration_required")
|
|
cases.append((
|
|
"blocked-observed-stage-model", blocked_cell, blocked_result, "blocked-stage-model",
|
|
lambda p: p["binding"]["effective_bindings"].__setitem__(
|
|
0, {"stage": "selector", "model": "gpth", "effort": "xhigh"}
|
|
),
|
|
))
|
|
cases.append((
|
|
"blocked-observed-stage-order", blocked_cell, blocked_result, "blocked-stage-order",
|
|
lambda p: p["binding"]["effective_bindings"].reverse(),
|
|
))
|
|
|
|
for label, cell, result, name, mutate in cases:
|
|
with self.subTest(label=label, name=name):
|
|
mutant_path = _write_mutant(cell, result, name, mutate)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, mutant_path, cell)
|
|
|
|
def test_reader_rejects_noncanonical_issue_order(self):
|
|
cell = _cell()
|
|
issues = (_issue("model_missing"), _issue("stream_incompatible"))
|
|
result = make_result(cell, _capability(), _requested_only(cell), issues)
|
|
self._write(self.root, "ordered.json", cell, result)
|
|
raw = (self.root / "ordered.json").read_text(encoding="ascii")
|
|
first = '{"code":"model_missing","resume_code":"register_model"}'
|
|
second = '{"code":"stream_incompatible","resume_code":"implement_stream_adapter"}'
|
|
self.assertIn(f"{first},{second}", raw)
|
|
(self.root / "swapped.json").write_text(raw.replace(f"{first},{second}", f"{second},{first}"), encoding="ascii")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "swapped.json", cell)
|
|
with self.assertRaises(ConnectivityValidationError):
|
|
make_result(cell, _capability(), _requested_only(cell), tuple(reversed(issues)))
|
|
|
|
def test_symlinked_roots_parents_and_targets_are_rejected(self):
|
|
outside = self.base / "outside"
|
|
outside.mkdir()
|
|
real_root = self.base / "real" / "evidence"
|
|
real_root.mkdir(parents=True)
|
|
(self.base / "link").symlink_to(self.base / "real", target_is_directory=True)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.base / "link" / "evidence", "preflight.json")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.base / "link" / "evidence", "preflight.json", self.cell)
|
|
self.assertFalse((real_root / "preflight.json").exists())
|
|
self._write(real_root, "preflight.json")
|
|
self.assertEqual(read_evidence(real_root, "preflight.json", self.cell)["status"], "ready")
|
|
|
|
(self.base / "root-link").symlink_to(self.root, target_is_directory=True)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.base / "root-link", "preflight.json")
|
|
|
|
(self.root / "nested").symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, "nested/preflight.json")
|
|
self.assertFalse((outside / "preflight.json").exists())
|
|
|
|
(self.root / "link.json").symlink_to(outside / "escape.json")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, "link.json")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "link.json", self.cell)
|
|
self.assertFalse((outside / "escape.json").exists())
|
|
|
|
def test_oversized_directory_and_non_regular_targets_are_rejected(self):
|
|
self._write(self.root, "small.json")
|
|
self.assertEqual(read_evidence(self.root, "small.json", self.cell)["status"], "ready")
|
|
(self.root / "big.json").write_bytes(b"{" + b" " * (MAX_EVIDENCE_BYTES + 16) + b"}")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "big.json", self.cell)
|
|
(self.root / "dir.json").mkdir()
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "dir.json", self.cell)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, "dir.json")
|
|
if hasattr(os, "mkfifo"):
|
|
os.mkfifo(self.root / "fifo.json")
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, "fifo.json", self.cell)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, "fifo.json")
|
|
|
|
def test_private_identity_and_escaping_paths_are_rejected(self):
|
|
for identity in ("https://private.invalid", "sha256:" + "z" * 64, "", "sha256:abcd", None):
|
|
with self.subTest(identity=identity):
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
canonical_evidence_bytes(self.cell, self.result, identity, self.config_identity)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
canonical_evidence_bytes(self.cell, self.result, self.endpoint_identity, identity)
|
|
for relative in ("../escape.json", "/abs.json", "nested/../escape.json", "preflight.txt", "preflight", "", ".json"):
|
|
with self.subTest(relative=relative):
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
self._write(self.root, relative)
|
|
with self.assertRaises(ConnectivityEvidenceError):
|
|
read_evidence(self.root, relative, self.cell)
|
|
self.assertFalse((self.base / "escape.json").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|