tegwick
4310c15eac
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a06ba0-10aa-7ea0-b20a-4f3fac39efe9
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from rein_aharness.close_outbox import CloseOutbox
|
|
from rein_aharness.ops_run_client import OpsRunConfig, OpsRunError
|
|
from rein_aharness.readiness import run_readiness_checks
|
|
|
|
|
|
class _Client:
|
|
def __init__(self, *, repo_map: dict[str, str] | None = None, error=None):
|
|
self.config = OpsRunConfig(repo_map=repo_map or {})
|
|
self.error = error
|
|
self.list_calls = 0
|
|
|
|
def list_open(self, *, limit: int = 50):
|
|
self.list_calls += 1
|
|
assert limit == 1
|
|
if self.error is not None:
|
|
raise self.error
|
|
return []
|
|
|
|
|
|
def test_readiness_is_read_only_and_passes_minimal_runtime(tmp_path: Path) -> None:
|
|
client = _Client()
|
|
report = run_readiness_checks(
|
|
client, # type: ignore[arg-type]
|
|
CloseOutbox(state_dir=tmp_path / "state"),
|
|
environ={},
|
|
)
|
|
|
|
assert report.ok
|
|
assert client.list_calls == 1
|
|
assert [check.name for check in report.checks] == [
|
|
"close_outbox",
|
|
"profiled_runtime",
|
|
"activity_core",
|
|
]
|
|
|
|
|
|
def test_readiness_fails_before_claim_when_queue_probe_is_unavailable(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
client = _Client(error=OpsRunError("transport", action="list"))
|
|
|
|
report = run_readiness_checks(
|
|
client, # type: ignore[arg-type]
|
|
CloseOutbox(state_dir=tmp_path / "state"),
|
|
environ={},
|
|
)
|
|
|
|
assert not report.ok
|
|
failure = next(check for check in report.checks if check.name == "activity_core")
|
|
assert failure.detail == "read-only queue probe failed (OpsRunError)"
|
|
|
|
|
|
def test_readiness_refuses_quarantine_and_missing_workspace(tmp_path: Path) -> None:
|
|
outbox = CloseOutbox(state_dir=tmp_path / "state")
|
|
(outbox.quarantine_dir / "entry.deadbeef.json").write_text("{}", encoding="utf-8")
|
|
client = _Client(repo_map={"missing": str(tmp_path / "absent")})
|
|
|
|
report = run_readiness_checks(
|
|
client, # type: ignore[arg-type]
|
|
outbox,
|
|
environ={},
|
|
check_activity_core=False,
|
|
)
|
|
|
|
assert not report.ok
|
|
failed = {check.name for check in report.checks if not check.ok}
|
|
assert failed == {"close_outbox", "repository:missing"}
|
|
|
|
|
|
def test_readiness_requires_exact_profile_versions_without_optional_imports(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
report = run_readiness_checks(
|
|
_Client(), # type: ignore[arg-type]
|
|
CloseOutbox(state_dir=tmp_path / "state"),
|
|
environ={"AGENT_HARNESS_REQUIRED_PROFILE_REFS": "harness.agent-dev-local"},
|
|
check_activity_core=False,
|
|
)
|
|
|
|
assert not report.ok
|
|
failure = next(check for check in report.checks if check.name.startswith("profile:"))
|
|
assert failure.detail == "production profile reference must pin an exact version"
|