1144 lines
40 KiB
Python
1144 lines
40 KiB
Python
"""Focused tests for Agent Hermes' direct Codex and Claude Kanban lanes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from contextlib import nullcontext
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
|
|
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
|
|
HERMES = Path(__file__).parents[2] / "services/hermes"
|
|
KEYCLOAK = Path(__file__).parents[2] / "services/keycloak"
|
|
FLUX_HERMES = (
|
|
Path(__file__).parents[2]
|
|
/ "clusters/atlas/flux-system/applications/hermes/kustomization.yaml"
|
|
)
|
|
|
|
|
|
def _load(name: str):
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
lanes = _load("cli_lane_runner")
|
|
policy = _load("claude_command_policy")
|
|
migration = _load("migrate_herdr_state")
|
|
auth_patch = _load("patch_hermes_auth")
|
|
tui_gateway_patch = _load("patch_tui_gateway")
|
|
codex_runtime_patch = _load("patch_codex_runtime")
|
|
ttyd_patch = _load("patch_ttyd_index")
|
|
client_config = _load("configure_agent_clients")
|
|
|
|
|
|
def _agent_deployment() -> dict:
|
|
return yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
|
|
|
|
def _oauth_deployment(name: str) -> dict:
|
|
documents = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
|
|
if item
|
|
]
|
|
return next(
|
|
item
|
|
for item in documents
|
|
if item["kind"] == "Deployment" and item["metadata"]["name"] == name
|
|
)
|
|
|
|
|
|
def test_auto_lane_always_uses_the_jetson_decision(tmp_path: Path):
|
|
router = tmp_path / "router.py"
|
|
router.write_text(
|
|
"""
|
|
class Decision:
|
|
shape = "review"
|
|
effort = "xhigh"
|
|
provider = "claude"
|
|
classifier = "jetson"
|
|
reason = "local vote"
|
|
latency_ms = 17
|
|
|
|
def classify_task(prompt):
|
|
assert "security review" in prompt
|
|
return Decision()
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
routes = tmp_path / "routes.json"
|
|
routes.write_text(
|
|
json.dumps(
|
|
{
|
|
"routes": {
|
|
"claude-xhigh": [
|
|
"anthropic/claude-opus-5",
|
|
"openai-codex/gpt-5.6-sol",
|
|
]
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
route = lanes.select_route(
|
|
"Perform the security review.",
|
|
"cli-auto",
|
|
routing_path=routes,
|
|
router_path=router,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.model == "claude-opus-5"
|
|
assert route.effort == "xhigh"
|
|
assert route.classifier == "jetson"
|
|
assert route.latency_ms == 17
|
|
|
|
|
|
def test_manual_lane_still_calls_jetson_before_applying_override(tmp_path: Path):
|
|
router = tmp_path / "router.py"
|
|
router.write_text(
|
|
"""
|
|
called = 0
|
|
class Decision:
|
|
shape = "question"
|
|
effort = "low"
|
|
provider = "codex"
|
|
classifier = "jetson"
|
|
reason = "local vote"
|
|
latency_ms = 8
|
|
def classify_task(prompt):
|
|
global called
|
|
called += 1
|
|
return Decision()
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
routes = tmp_path / "routes.json"
|
|
routes.write_text(
|
|
json.dumps({"routes": {"claude-high": ["anthropic/claude-opus-5"]}}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
route = lanes.select_route(
|
|
"Implement it.",
|
|
"cli-claude-high",
|
|
routing_path=routes,
|
|
router_path=router,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.effort == "high"
|
|
assert "manual lane override" in route.reason
|
|
|
|
|
|
def test_cross_provider_retry_excludes_failed_provider(tmp_path: Path):
|
|
router = tmp_path / "router.py"
|
|
router.write_text(
|
|
"""
|
|
class Decision:
|
|
shape = "implementation"
|
|
effort = "high"
|
|
provider = "codex"
|
|
classifier = "jetson"
|
|
reason = "retry vote"
|
|
latency_ms = 9
|
|
def classify_task(prompt): return Decision()
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
routes = tmp_path / "routes.json"
|
|
routes.write_text(
|
|
json.dumps({"routes": {"claude-high": ["anthropic/claude-opus-5"]}}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
route = lanes.select_route(
|
|
"Retry after capacity exhaustion.",
|
|
"cli-auto",
|
|
exclude_provider="codex",
|
|
routing_path=routes,
|
|
router_path=router,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.classifier == "jetson"
|
|
|
|
|
|
def test_claude_session_is_reserved_before_first_process(tmp_path: Path, monkeypatch):
|
|
state: dict = {}
|
|
state_file = tmp_path / "state.json"
|
|
observed = {}
|
|
|
|
def fake_stream(command, **kwargs):
|
|
observed["command"] = command
|
|
observed["state"] = json.loads(state_file.read_text())
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
|
|
lanes.run_provider(
|
|
route,
|
|
"Do the work.",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "worker.log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
|
|
reserved = observed["state"]["claude_session_id"]
|
|
assert reserved
|
|
assert ["--session-id", reserved] == observed["command"][-4:-2]
|
|
|
|
|
|
def test_claude_recovery_uses_verified_error_state_machine(tmp_path: Path, monkeypatch):
|
|
state = {"claude_session_id": "d4f91f59-88cb-43d0-a64f-f3284688fe9e", "claude_started": True}
|
|
state_file = tmp_path / "state.json"
|
|
calls = []
|
|
|
|
def fake_stream(command, **kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(
|
|
1,
|
|
"No conversation found with session ID: d4f91f59-88cb-43d0-a64f-f3284688fe9e",
|
|
None,
|
|
False,
|
|
)
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
|
|
|
|
assert "--resume" in calls[0]
|
|
assert "--session-id" in calls[1]
|
|
assert calls[0][calls[0].index("--resume") + 1] == calls[1][calls[1].index("--session-id") + 1]
|
|
|
|
|
|
def test_claude_collision_never_double_starts_a_reserved_session(tmp_path: Path, monkeypatch):
|
|
state = {"claude_session_id": "24927cee-f3da-4bcf-b624-602c22860155"}
|
|
state_file = tmp_path / "state.json"
|
|
calls = []
|
|
|
|
def fake_stream(command, **kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(1, "Session ID already in use", None, False)
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
|
|
|
|
assert "--session-id" in calls[0]
|
|
assert "--resume" in calls[1]
|
|
assert calls[0][calls[0].index("--session-id") + 1] == calls[1][calls[1].index("--resume") + 1]
|
|
|
|
|
|
def test_codex_thread_started_is_persisted_before_completion(tmp_path: Path):
|
|
state: dict = {}
|
|
state_file = tmp_path / "state.json"
|
|
|
|
lanes._event_payload(
|
|
"codex",
|
|
json.dumps({"type": "thread.started", "thread_id": "thread-123"}),
|
|
state,
|
|
state_file,
|
|
)
|
|
|
|
assert json.loads(state_file.read_text())["codex_thread_id"] == "thread-123"
|
|
|
|
|
|
def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
|
|
result = lanes.stream_process(
|
|
[sys.executable, "-c", "print('authentication work completed')"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "worker.log",
|
|
state={},
|
|
state_file=tmp_path / "state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.capacity_failure is False
|
|
|
|
|
|
def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch):
|
|
task = SimpleNamespace(id="t_auto", assignee=None, status="ready")
|
|
assigned = []
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
def assign_task(_conn, task_id, profile):
|
|
assigned.append((task_id, profile))
|
|
task.assignee = profile
|
|
return True
|
|
|
|
fake_db = SimpleNamespace(
|
|
list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(),
|
|
recompute_ready=lambda _conn: None,
|
|
list_tasks=lambda _conn: [task],
|
|
assign_task=assign_task,
|
|
get_task=lambda _conn, _task_id: task,
|
|
claim_task=lambda _conn, _task_id, **_kwargs: task,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
|
|
assert lanes.claim_ready(set(), 1) == [("cassandra", "t_auto")]
|
|
assert assigned == [("t_auto", "cli-auto")]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("result", "expected_action"),
|
|
[
|
|
(lanes.ProcessResult(0, "plain text only", None, False), "block"),
|
|
(
|
|
lanes.ProcessResult(
|
|
0,
|
|
"",
|
|
{
|
|
"status": "completed",
|
|
"summary": "done",
|
|
"changed_files": ["src/a.py"],
|
|
"tests_run": ["pytest -q"],
|
|
"artifacts": ["reports/result.json"],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
),
|
|
"complete",
|
|
),
|
|
],
|
|
)
|
|
def test_claim_requires_structured_evidence_and_surfaces_artifacts(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
result,
|
|
expected_action,
|
|
):
|
|
task = SimpleNamespace(
|
|
id="t_worker",
|
|
current_run_id=4,
|
|
assignee="cli-auto",
|
|
max_runtime_seconds=60,
|
|
)
|
|
calls = []
|
|
heartbeats = []
|
|
artifact = tmp_path / "reports/result.json"
|
|
artifact.parent.mkdir()
|
|
artifact.write_text("{}\n", encoding="utf-8")
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
fake_db = SimpleNamespace(
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(),
|
|
get_task=lambda _conn, _task_id: task,
|
|
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
|
|
_resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_worker"),
|
|
set_branch_name=lambda *_args: None,
|
|
set_workspace_path=lambda *_args: None,
|
|
build_worker_context=lambda *_args: "bounded objective",
|
|
heartbeat_worker=lambda _conn, _task_id, *, note, expected_run_id: (
|
|
heartbeats.append((note, expected_run_id)) or True
|
|
),
|
|
add_comment=lambda *_args: None,
|
|
complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
|
|
block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"select_route",
|
|
lambda *_args, **_kwargs: lanes.Route(
|
|
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
|
|
),
|
|
)
|
|
def run_provider(*args, **_kwargs):
|
|
assert args[6]("working") is True
|
|
return result
|
|
|
|
monkeypatch.setattr(lanes, "run_provider", run_provider)
|
|
|
|
lanes.execute_claim("cassandra", "t_worker")
|
|
|
|
assert calls[0][0] == expected_action
|
|
assert heartbeats == [("working", 4)]
|
|
if expected_action == "complete":
|
|
assert calls[0][1]["metadata"]["artifacts"] == [str(artifact)]
|
|
assert calls[0][1]["metadata"]["tests_run"] == ["pytest -q"]
|
|
else:
|
|
assert calls[0][1]["kind"] == "capability"
|
|
|
|
|
|
def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch):
|
|
task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto")
|
|
calls = []
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
fake_db = SimpleNamespace(
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(),
|
|
get_task=lambda _conn, _task_id: task,
|
|
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
|
|
_resolve_worktree_workspace=lambda _task, board: (_ for _ in ()).throw(
|
|
ValueError("not a Git repository")
|
|
),
|
|
block_task=lambda *_args, **kwargs: calls.append(kwargs),
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
|
|
|
|
lanes.execute_claim("cassandra", "t_bad_worktree")
|
|
|
|
assert calls[0]["kind"] == "capability"
|
|
assert calls[0]["expected_run_id"] == 7
|
|
assert "not a Git repository" in calls[0]["reason"]
|
|
|
|
|
|
def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path):
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
inside = workspace / "report.json"
|
|
outside = tmp_path / "auth.json"
|
|
inside.write_text("{}\n", encoding="utf-8")
|
|
outside.write_text("secret\n", encoding="utf-8")
|
|
|
|
assert lanes.workspace_artifacts(
|
|
workspace,
|
|
["report.json", str(outside), "missing.json"],
|
|
) == [str(inside)]
|
|
|
|
|
|
def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: Path, monkeypatch):
|
|
task = SimpleNamespace(
|
|
id="t_resume",
|
|
current_run_id=9,
|
|
assignee="cli-auto",
|
|
max_runtime_seconds=60,
|
|
)
|
|
state_file = tmp_path / "state.json"
|
|
state_file.write_text(
|
|
json.dumps({"current_route": {"provider": "claude"}}),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "worker.log").write_text("prior provider evidence", encoding="utf-8")
|
|
prompts = []
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
fake_db = SimpleNamespace(
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(),
|
|
get_task=lambda _conn, _task_id: task,
|
|
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
|
|
_resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_resume"),
|
|
set_branch_name=lambda *_args: None,
|
|
set_workspace_path=lambda *_args: None,
|
|
build_worker_context=lambda *_args: "resume objective",
|
|
add_comment=lambda *_args: None,
|
|
complete_task=lambda *_args, **_kwargs: None,
|
|
block_task=lambda *_args, **_kwargs: None,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file)
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"select_route",
|
|
lambda *_args, **_kwargs: lanes.Route(
|
|
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"git_handoff",
|
|
lambda _workspace, output: f"HANDOFF:{output}",
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"run_provider",
|
|
lambda _route, prompt, *_args, **_kwargs: (
|
|
prompts.append(prompt)
|
|
or lanes.ProcessResult(
|
|
0,
|
|
"",
|
|
{
|
|
"status": "completed",
|
|
"summary": "done",
|
|
"changed_files": [],
|
|
"tests_run": [],
|
|
"artifacts": [],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
)
|
|
),
|
|
)
|
|
|
|
lanes.execute_claim("cassandra", "t_resume")
|
|
|
|
assert "HANDOFF:prior provider evidence" in prompts[0]
|
|
|
|
|
|
def test_provider_commands_are_structured_unattended_and_capped(tmp_path: Path):
|
|
route = lanes.Route("codex", "gpt-5.6-sol", "xhigh", "codex-xhigh", "jetson", "vote", 1, ())
|
|
command = lanes._codex_command(route, "Work.", tmp_path, {}, tmp_path / "result.json")
|
|
assert "--dangerously-bypass-approvals-and-sandbox" in command
|
|
assert "--json" in command
|
|
assert "--output-schema" in command
|
|
assert 'model_reasoning_effort="xhigh"' in command
|
|
|
|
claude_state = {"claude_session_id": "13864642-2985-4f91-bef5-53f145f878e8"}
|
|
claude = lanes._claude_command(
|
|
lanes.Route("claude", "claude-opus-5", "xhigh", "claude-xhigh", "jetson", "vote", 1, ()),
|
|
"Review.",
|
|
claude_state,
|
|
False,
|
|
)
|
|
assert "--dangerously-skip-permissions" in claude
|
|
assert "--output-format" in claude and "stream-json" in claude
|
|
assert "--json-schema" in claude
|
|
assert "--disallowedTools" in claude
|
|
assert "Bash(kubectl apply *)" not in claude
|
|
assert "Bash(flux reconcile *)" not in claude
|
|
assert "max" not in claude
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"git push --force origin main",
|
|
"git reset --hard HEAD~1",
|
|
"git clean -fd",
|
|
],
|
|
)
|
|
def test_claude_pretool_hook_blocks_hard_denies(command: str):
|
|
assert policy.denial_reason(command)
|
|
|
|
|
|
def test_claude_pretool_hook_allows_normal_engineering():
|
|
assert policy.denial_reason("pytest -q testing/tests") is None
|
|
assert policy.denial_reason("git push origin feature/hermes") is None
|
|
assert policy.denial_reason("kubectl delete pod -n cassandra stuck-worker") is None
|
|
assert policy.denial_reason("flux reconcile kustomization hermes") is None
|
|
assert policy.denial_reason("vault kv get kv/atlas/hermes") is None
|
|
|
|
|
|
def test_claude_settings_preserve_state_and_install_three_guardrail_layers(tmp_path: Path):
|
|
state = tmp_path / ".claude.json"
|
|
settings = tmp_path / "settings.json"
|
|
state.write_text('{"promptQueueUseCount": 4}\n', encoding="utf-8")
|
|
settings.write_text(
|
|
json.dumps(
|
|
{
|
|
"theme": "dark",
|
|
"permissions": {
|
|
"deny": [
|
|
"Bash(kubectl apply *)",
|
|
"Bash(flux reconcile *)",
|
|
"Bash(vault kv *)",
|
|
"Bash(custom-owner-rule *)",
|
|
]
|
|
},
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
client_config.configure_claude_state(state)
|
|
client_config.configure_claude_settings(settings)
|
|
|
|
state_value = json.loads(state.read_text())
|
|
settings_value = json.loads(settings.read_text())
|
|
assert state_value["promptQueueUseCount"] == 4
|
|
assert state_value["bypassPermissionsModeAccepted"] is True
|
|
assert settings_value["theme"] == "dark"
|
|
assert "Bash(git reset --hard *)" in settings_value["permissions"]["deny"]
|
|
assert "Bash(custom-owner-rule *)" in settings_value["permissions"]["deny"]
|
|
assert "Bash(kubectl apply *)" not in settings_value["permissions"]["deny"]
|
|
assert "Bash(flux reconcile *)" not in settings_value["permissions"]["deny"]
|
|
assert "Bash(vault kv *)" not in settings_value["permissions"]["deny"]
|
|
hook = settings_value["hooks"]["PreToolUse"][0]["hooks"][0]
|
|
assert "claude_command_policy.py" in hook["command"]
|
|
|
|
|
|
def test_legacy_state_is_archived_without_removing_provider_transcripts(tmp_path: Path):
|
|
session = tmp_path / "home/.config/herdr/session.json"
|
|
session.parent.mkdir(parents=True)
|
|
session.write_text('{"agents":[{"agent":"claude","session_id":"abc"}]}', encoding="utf-8")
|
|
binary = tmp_path / "tools/bin/herdr"
|
|
binary.parent.mkdir(parents=True)
|
|
binary.write_text("legacy", encoding="utf-8")
|
|
(tmp_path / "home/.claude").mkdir()
|
|
(tmp_path / "home/.codex").mkdir()
|
|
|
|
archive = migration.archive_legacy_state(tmp_path)
|
|
|
|
value = json.loads(archive.read_text())
|
|
assert value["legacy_session"]["agents"][0]["session_id"] == "abc"
|
|
assert (tmp_path / "home/.claude").is_dir()
|
|
assert (tmp_path / "home/.codex").is_dir()
|
|
assert not binary.exists()
|
|
assert not (tmp_path / "home/.config/herdr").exists()
|
|
|
|
|
|
def test_agent_uses_one_native_kanban_control_plane():
|
|
configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
assert config["kanban"]["dispatch_in_gateway"] is True
|
|
assert config["kanban"]["default_assignee"] == "cli-auto"
|
|
assert config["plugins"]["enabled"] == ["auto-router"]
|
|
|
|
deployment = _agent_deployment()
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
names = {item["name"] for item in pod["containers"]}
|
|
assert "cli-lane-runner" in names
|
|
assert "terminal" in names
|
|
assert not any("herdr" in name for name in names)
|
|
rendered = (HERMES / "agent-deployment.yaml").read_text()
|
|
assert "herdr server" not in rendered
|
|
assert "herdr-dispatch" not in rendered
|
|
|
|
|
|
def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path():
|
|
deployment = _agent_deployment()
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
containers = {item["name"]: item for item in pod["containers"]}
|
|
assert "webui" not in containers
|
|
|
|
assert "dashboard" not in containers
|
|
hermes = containers["hermes"]
|
|
assert hermes["command"] == ["/init", "/opt/hermes/docker/main-wrapper.sh"]
|
|
assert hermes["args"] == ["gateway", "run"]
|
|
hermes_env = {item["name"]: item["value"] for item in hermes["env"]}
|
|
assert hermes_env["HERMES_DASHBOARD"] == "1"
|
|
assert hermes_env["HERMES_DASHBOARD_HOST"] == "127.0.0.1"
|
|
assert hermes_env["HERMES_DASHBOARD_PORT"] == "9119"
|
|
assert hermes_env["HERMES_TUI_AGENT_INIT_TIMEOUT_S"] == "180"
|
|
assert hermes["securityContext"]["runAsUser"] == 0
|
|
assert hermes["securityContext"]["runAsGroup"] == 0
|
|
for probe_name in ("startupProbe", "readinessProbe", "livenessProbe"):
|
|
probe = hermes[probe_name]
|
|
assert probe["exec"]["command"] == [
|
|
"curl",
|
|
"-fsS",
|
|
"http://127.0.0.1:9119/api/status",
|
|
]
|
|
|
|
terminal = containers["terminal"]
|
|
command = terminal["args"][0]
|
|
assert "--base-path /terminal" in command
|
|
assert "--check-origin" not in command
|
|
assert "/usr/bin/tmux new-session -A" in command
|
|
assert "--continue" in command
|
|
assert "--yolo" in command
|
|
terminal_env = {item["name"]: item["value"] for item in terminal["env"]}
|
|
assert terminal_env["HERMES_TUI_AGENT_INIT_TIMEOUT_S"] == "180"
|
|
|
|
args = containers["oauth2-proxy"]["args"]
|
|
terminal_upstream = "--upstream=http://127.0.0.1:7681/terminal/"
|
|
dashboard_upstream = "--upstream=http://127.0.0.1:9119/"
|
|
assert terminal_upstream in args
|
|
assert dashboard_upstream in args
|
|
assert args.index(terminal_upstream) < args.index(dashboard_upstream)
|
|
assert "--pass-host-header=false" in args
|
|
assert "--cookie-refresh=19m" in args
|
|
assert "--session-store-type=redis" in args
|
|
assert any(
|
|
arg.startswith("--redis-connection-url=redis://hermes-oauth-sessions.")
|
|
for arg in args
|
|
)
|
|
|
|
patch_init = next(
|
|
item for item in pod["initContainers"]
|
|
if item["name"] == "patch-tui-gateway"
|
|
)
|
|
assert patch_init["command"][-1] == "/patched/server.py"
|
|
for name in ("hermes", "terminal"):
|
|
mounts = containers[name]["volumeMounts"]
|
|
assert {
|
|
"name": "tui-gateway-patch",
|
|
"mountPath": "/opt/hermes/tui_gateway/server.py",
|
|
"subPath": "server.py",
|
|
} in mounts
|
|
|
|
ingress_documents = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "agent-ingress.yaml").read_text())
|
|
if item
|
|
]
|
|
middlewares = {
|
|
item["metadata"]["name"]: item
|
|
for item in ingress_documents
|
|
if item["kind"] == "Middleware"
|
|
}
|
|
assert middlewares["hermes-agent-terminal-slash"]["spec"]["redirectRegex"][
|
|
"replacement"
|
|
].endswith("/terminal/")
|
|
assert middlewares["hermes-agent-stock-dashboard-headers"]["spec"]["headers"][
|
|
"customRequestHeaders"
|
|
]["Origin"] == "http://127.0.0.1:9119"
|
|
ingresses = {
|
|
item["metadata"]["name"]: item
|
|
for item in ingress_documents
|
|
if item["kind"] == "Ingress"
|
|
}
|
|
assert ingresses["hermes-agent-dashboard"]["metadata"]["annotations"][
|
|
"traefik.ingress.kubernetes.io/router.middlewares"
|
|
] == "hermes-hermes-agent-stock-dashboard-headers@kubernetescrd"
|
|
assert ingresses["hermes-agent-terminal"]["metadata"]["annotations"][
|
|
"traefik.ingress.kubernetes.io/router.middlewares"
|
|
] == "hermes-hermes-agent-terminal-slash@kubernetescrd"
|
|
|
|
|
|
def test_agent_dashboard_reconnects_all_transient_websockets():
|
|
dockerfile = (
|
|
HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent"
|
|
).read_text(encoding="utf-8")
|
|
assert "eventsRetryAttempt.current" in dockerfile
|
|
assert "if (!unmounting) setVersion((v) => v + 1);" in dockerfile
|
|
assert "events feed rejected (${ev.code}) — reload the page" in dockerfile
|
|
assert 'url = await api.buildWsUrl("/api/pty", params);' in dockerfile
|
|
assert 'url = await buildWsUrl("/api/events", { channel });' in dockerfile
|
|
|
|
|
|
def test_agent_refreshes_routes_after_restoring_cli_logins():
|
|
deployment = _agent_deployment()
|
|
init_containers = {
|
|
item["name"]: item
|
|
for item in deployment["spec"]["template"]["spec"]["initContainers"]
|
|
}
|
|
configure = init_containers["configure-agent-clients"]
|
|
command = configure["command"][-1]
|
|
assert "configure_agent_clients.py" in command
|
|
assert command.index("configure_agent_clients.py") < command.index(
|
|
"hermes_coordinator.py --once"
|
|
)
|
|
env = {item["name"]: item["value"] for item in configure["env"]}
|
|
assert env["HERMES_AUTH_FILE"] == "/shared-auth/auth.json"
|
|
assert env["PYTHONPATH"] == "/opt/hermes"
|
|
for name in ("bootstrap-coordinator", "configure-agent-clients"):
|
|
route_env = {
|
|
item["name"]: item["value"]
|
|
for item in init_containers[name]["env"]
|
|
}
|
|
assert route_env["CODEX_HOME"] == "/opt/data/home/.codex"
|
|
assert "/opt/data/tools/bin" in route_env["PATH"]
|
|
|
|
containers = {
|
|
item["name"]: item
|
|
for item in deployment["spec"]["template"]["spec"]["containers"]
|
|
}
|
|
steward_env = {
|
|
item["name"]: item["value"]
|
|
for item in containers["model-steward"]["env"]
|
|
}
|
|
assert steward_env["CODEX_HOME"] == "/opt/data/home/.codex"
|
|
assert "/opt/data/tools/bin" in steward_env["PATH"]
|
|
|
|
|
|
def test_flux_health_checks_follow_the_owner_oauth_sidecar():
|
|
flux = yaml.safe_load(FLUX_HERMES.read_text())
|
|
checks = {
|
|
(item["kind"], item["name"])
|
|
for item in flux["spec"]["healthChecks"]
|
|
}
|
|
assert ("Deployment", "hermes-agent") in checks
|
|
assert ("Deployment", "oauth2-proxy-hermes-agent") not in checks
|
|
|
|
|
|
def test_agent_auth_is_bstein_group_and_email_bounded():
|
|
deployment = _agent_deployment()
|
|
oauth = next(
|
|
item for item in deployment["spec"]["template"]["spec"]["containers"]
|
|
if item["name"] == "oauth2-proxy"
|
|
)
|
|
args = oauth["args"]
|
|
assert "--user-id-claim=sub" in args
|
|
assert "--oidc-groups-claim=groups" in args
|
|
assert "--allowed-group=/hermes-owner" in args
|
|
assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in args
|
|
script = (KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh").read_text()
|
|
assert 'group_name="hermes-owner"' in script
|
|
assert "username=bstein&exact=true" in script
|
|
assert '"full.path":"true"' in script
|
|
|
|
|
|
def test_agent_network_boundary_allows_only_authenticated_web_and_broker_surfaces():
|
|
documents = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
|
|
if item
|
|
]
|
|
isolation = next(item for item in documents if item.get("metadata", {}).get("name") == "hermes-agent-isolation")
|
|
assert isolation["spec"]["ingress"] == [
|
|
{
|
|
"from": [
|
|
{
|
|
"namespaceSelector": {
|
|
"matchLabels": {
|
|
"kubernetes.io/metadata.name": "traefik"
|
|
}
|
|
},
|
|
"podSelector": {
|
|
"matchLabels": {"app.kubernetes.io/name": "traefik"}
|
|
},
|
|
}
|
|
],
|
|
"ports": [{"protocol": "TCP", "port": 4180}],
|
|
},
|
|
{
|
|
"from": [
|
|
{
|
|
"podSelector": {
|
|
"matchLabels": {"app": "hermes-chat-tenant"}
|
|
}
|
|
}
|
|
],
|
|
"ports": [
|
|
{"protocol": "TCP", "port": 9002},
|
|
{"protocol": "TCP", "port": 9003},
|
|
],
|
|
},
|
|
]
|
|
assert isolation["spec"]["egress"] == [{}]
|
|
|
|
|
|
def test_owner_agent_has_cluster_admin_kubernetes_context():
|
|
config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
|
|
assert config["current-context"] == "atlas-owner"
|
|
assert config["contexts"][0]["context"]["namespace"] == "default"
|
|
rbac_path = HERMES / "agent-rbac.yaml"
|
|
documents = [item for item in yaml.safe_load_all(rbac_path.read_text()) if item]
|
|
binding = next(item for item in documents if item["kind"] == "ClusterRoleBinding")
|
|
assert binding["roleRef"] == {
|
|
"apiGroup": "rbac.authorization.k8s.io",
|
|
"kind": "ClusterRole",
|
|
"name": "cluster-admin",
|
|
}
|
|
assert binding["subjects"] == [
|
|
{"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
|
|
]
|
|
|
|
|
|
def test_owner_agent_installs_the_pinned_operator_toolchain():
|
|
script = (SCRIPTS / "install_agent_tools.sh").read_text()
|
|
for value in [
|
|
"flux",
|
|
"helm",
|
|
"kustomize",
|
|
"jq",
|
|
"yq",
|
|
"gh",
|
|
"vault",
|
|
"sops",
|
|
"age",
|
|
"age-keygen",
|
|
"k9s",
|
|
"terraform",
|
|
]:
|
|
assert value in script
|
|
assert script.count("sha256sum -c -") == 1
|
|
|
|
deployment = _agent_deployment()
|
|
installer = next(
|
|
item
|
|
for item in deployment["spec"]["template"]["spec"]["initContainers"]
|
|
if item["name"] == "install-agent-tools"
|
|
)
|
|
assert "/bin/sh /opt/coordinator/install_agent_tools.sh" in installer["command"][2]
|
|
assert any(mount["name"] == "coordinator" for mount in installer["volumeMounts"])
|
|
|
|
|
|
def test_owner_agent_uses_only_the_canonical_hostname():
|
|
paths = [
|
|
HERMES / "agent-configmap.yaml",
|
|
HERMES / "agent-deployment.yaml",
|
|
HERMES / "agent-ingress.yaml",
|
|
Path(__file__).parents[2] / "scripts/ops/hermes_triage_monitor.py",
|
|
]
|
|
for path in paths:
|
|
content = path.read_text()
|
|
assert "agent.bstein.dev" not in content
|
|
assert "agent.hermes.bstein.dev" in content
|
|
|
|
|
|
def test_agent_reconnect_retains_complete_history_and_long_tool_budget():
|
|
configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
display = config["display"]
|
|
assert display["resume_exchanges"] >= 10000
|
|
assert display["resume_max_user_chars"] >= 10000000
|
|
assert display["resume_max_assistant_chars"] >= 10000000
|
|
assert config["agent"]["max_turns"] == 180
|
|
assert config["delegation"]["max_iterations"] == 120
|
|
|
|
|
|
def test_auth_patch_honors_explicit_shared_store(tmp_path: Path):
|
|
source = tmp_path / "auth.py"
|
|
destination = tmp_path / "patched/auth.py"
|
|
source.write_text(
|
|
'from pathlib import Path\nimport os\n\ndef _auth_file_path() -> Path:\n path = get_hermes_home() / "auth.json"\n return path\n',
|
|
encoding="utf-8",
|
|
)
|
|
auth_patch.patch(source, destination)
|
|
content = destination.read_text()
|
|
assert 'os.environ.get("HERMES_AUTH_FILE"' in content
|
|
|
|
|
|
def test_auth_patch_fails_closed_on_upstream_drift(tmp_path: Path):
|
|
source = tmp_path / "auth.py"
|
|
source.write_text("def changed():\n pass\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
auth_patch.patch(source, tmp_path / "patched.py")
|
|
|
|
|
|
def test_codex_runtime_patch_uses_cli_and_forwards_route(tmp_path: Path):
|
|
provider = tmp_path / "runtime_provider.py"
|
|
provider.write_text(codex_runtime_patch.PROVIDER_BEFORE, encoding="utf-8")
|
|
provider_out = tmp_path / "patched/runtime_provider.py"
|
|
codex_runtime_patch.patch_provider(provider, provider_out)
|
|
assert '"api_mode": "codex_app_server"' in provider_out.read_text()
|
|
|
|
session = tmp_path / "codex_app_server_session.py"
|
|
session.write_text(
|
|
codex_runtime_patch.SESSION_SIGNATURE_BEFORE
|
|
+ codex_runtime_patch.SESSION_REQUEST_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
session_out = tmp_path / "patched/codex_app_server_session.py"
|
|
codex_runtime_patch.patch_session(session, session_out)
|
|
session_content = session_out.read_text()
|
|
assert 'turn_params["model"] = model' in session_content
|
|
assert 'turn_params["effort"] = effort' in session_content
|
|
assert '"approvalPolicy": "never"' in session_content
|
|
assert '"sandboxPolicy": {"type": "dangerFullAccess"}' in session_content
|
|
|
|
turn = tmp_path / "codex_runtime.py"
|
|
turn.write_text(
|
|
codex_runtime_patch.FALLBACK_CONTEXT_BEFORE
|
|
+ codex_runtime_patch.TURN_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
turn_out = tmp_path / "patched/codex_runtime.py"
|
|
codex_runtime_patch.patch_turn(turn, turn_out)
|
|
turn_content = turn_out.read_text()
|
|
assert "model=str(getattr(agent" in turn_content
|
|
assert "build_cross_provider_codex_prompt" in turn_content
|
|
|
|
fallback = tmp_path / "chat_completion_helpers.py"
|
|
fallback.write_text(
|
|
codex_runtime_patch.FALLBACK_RESOLUTION_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
fallback_out = tmp_path / "patched/chat_completion_helpers.py"
|
|
codex_runtime_patch.patch_fallback(fallback, fallback_out)
|
|
fallback_content = fallback_out.read_text()
|
|
assert 'agent.api_mode = "codex_app_server"' in fallback_content
|
|
assert "agent._codex_cross_provider_fallback = True" in fallback_content
|
|
|
|
loop = tmp_path / "conversation_loop.py"
|
|
loop.write_text(
|
|
codex_runtime_patch.FALLBACK_DISPATCH_BEFORE
|
|
+ codex_runtime_patch.RETRY_FALLBACK_DISPATCH_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
loop_out = tmp_path / "patched/conversation_loop.py"
|
|
codex_runtime_patch.patch_loop(loop, loop_out)
|
|
loop_content = loop_out.read_text()
|
|
assert loop_content.count('if agent.api_mode == "codex_app_server"') == 2
|
|
assert "build_cross_provider_codex_prompt" in loop_content
|
|
retry_dispatch = loop_content.index(
|
|
"Fallback activation happens inside this retry loop"
|
|
)
|
|
api_kwargs = loop_content.find("agent._build_api_kwargs", retry_dispatch)
|
|
assert api_kwargs == -1 or retry_dispatch < api_kwargs
|
|
|
|
auxiliary = tmp_path / "auxiliary_client.py"
|
|
auxiliary.write_text(
|
|
codex_runtime_patch.AUXILIARY_TOKEN_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
auxiliary_out = tmp_path / "patched/auxiliary_client.py"
|
|
codex_runtime_patch.patch_auxiliary(auxiliary, auxiliary_out)
|
|
auxiliary_content = auxiliary_out.read_text()
|
|
assert 'os.environ.get("CODEX_HOME"' in auxiliary_content
|
|
assert 'Path(codex_home).expanduser() / "auth.json"' in auxiliary_content
|
|
assert "sole owner" in auxiliary_content
|
|
|
|
|
|
def test_agent_mounts_codex_auxiliary_runtime_patch():
|
|
deployment = _agent_deployment()
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
patch_init = next(
|
|
item for item in pod["initContainers"]
|
|
if item["name"] == "patch-codex-runtime"
|
|
)
|
|
assert patch_init["command"][-2:] == [
|
|
"/opt/hermes/agent/auxiliary_client.py",
|
|
"/patched/auxiliary_client.py",
|
|
]
|
|
expected_mount = {
|
|
"name": "codex-runtime-patch",
|
|
"mountPath": "/opt/hermes/agent/auxiliary_client.py",
|
|
"subPath": "auxiliary_client.py",
|
|
}
|
|
containers = {item["name"]: item for item in pod["containers"]}
|
|
for name in ("hermes", "terminal"):
|
|
assert expected_mount in containers[name]["volumeMounts"]
|
|
|
|
|
|
def test_codex_auxiliary_patch_reads_cli_token_without_copying_it(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
codex_home = tmp_path / ".codex"
|
|
codex_home.mkdir()
|
|
(codex_home / "auth.json").write_text(
|
|
json.dumps({"tokens": {"access_token": "cli-access-token"}}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
|
|
|
source = tmp_path / "auxiliary_client.py"
|
|
source.write_text(
|
|
"import json, logging, os, time\n"
|
|
"from pathlib import Path\n"
|
|
"logger = logging.getLogger(__name__)\n"
|
|
"def read_token():\n"
|
|
" try:\n"
|
|
" raise RuntimeError('Hermes provider store intentionally empty')\n"
|
|
+ codex_runtime_patch.AUXILIARY_TOKEN_BEFORE,
|
|
encoding="utf-8",
|
|
)
|
|
destination = tmp_path / "patched/auxiliary_client.py"
|
|
codex_runtime_patch.patch_auxiliary(source, destination)
|
|
namespace: dict = {}
|
|
exec(compile(destination.read_text(), str(destination), "exec"), namespace)
|
|
|
|
assert namespace["read_token"]() == "cli-access-token"
|
|
|
|
|
|
def test_codex_runtime_patch_fails_closed_on_upstream_drift(tmp_path: Path):
|
|
source = tmp_path / "runtime_provider.py"
|
|
source.write_text("def changed():\n pass\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
codex_runtime_patch.patch_provider(source, tmp_path / "patched.py")
|
|
|
|
|
|
def test_codex_runtime_migration_uses_owner_unsafe_mode(tmp_path: Path):
|
|
config = tmp_path / "config.yaml"
|
|
config.write_text("model: {}\n", encoding="utf-8")
|
|
codex_home = tmp_path / ".codex"
|
|
calls = []
|
|
|
|
class Report:
|
|
errors = []
|
|
|
|
@staticmethod
|
|
def summary():
|
|
return "configured"
|
|
|
|
def migrate(value, **kwargs):
|
|
calls.append((value, kwargs))
|
|
return Report()
|
|
|
|
client_config.configure_codex_runtime(config, migrate, codex_home)
|
|
|
|
assert calls[0][1]["default_permission_profile"] is None
|
|
assert calls[0][1]["codex_home"] == codex_home
|
|
content = (codex_home / "config.toml").read_text(encoding="utf-8")
|
|
assert 'approval_policy = "never"' in content
|
|
assert 'sandbox_mode = "danger-full-access"' in content
|
|
assert "default_permissions" not in content
|
|
|
|
|
|
def test_codex_owner_permissions_replace_stale_profile(tmp_path: Path):
|
|
config = tmp_path / "config.toml"
|
|
config.write_text(
|
|
'default_permissions = ":danger-no-sandbox"\n\n[features]\nhooks = true\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
client_config.configure_codex_owner_permissions(config)
|
|
client_config.configure_codex_owner_permissions(config)
|
|
|
|
content = config.read_text(encoding="utf-8")
|
|
assert content.count(client_config.OWNER_PERMISSIONS_BEGIN) == 1
|
|
assert content.count('approval_policy = "never"') == 1
|
|
assert content.count('sandbox_mode = "danger-full-access"') == 1
|
|
assert "default_permissions" not in content
|
|
assert "[features]\nhooks = true" in content
|
|
|
|
|
|
def test_tui_gateway_patch_extends_and_bounds_agent_startup(tmp_path: Path):
|
|
source = tmp_path / "server.py"
|
|
destination = tmp_path / "patched/server.py"
|
|
source.write_text(
|
|
"import os\n\n" + tui_gateway_patch.BEFORE + "\ndef unchanged():\n pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
tui_gateway_patch.patch(source, destination)
|
|
|
|
content = destination.read_text(encoding="utf-8")
|
|
assert "HERMES_TUI_AGENT_INIT_TIMEOUT_S" in content
|
|
assert 'configured = 180.0' in content
|
|
assert "return max(30.0, min(configured, 900.0))" in content
|
|
assert "timeout: float | None = None" in content
|
|
assert "ready.wait(timeout=wait_timeout)" in content
|
|
assert "def unchanged():" in content
|
|
|
|
|
|
def test_tui_gateway_patch_fails_closed_on_upstream_drift(tmp_path: Path):
|
|
source = tmp_path / "server.py"
|
|
source.write_text("def changed():\n pass\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
tui_gateway_patch.patch(source, tmp_path / "patched.py")
|
|
|
|
|
|
def test_ttyd_clipboard_and_reconnect_patch_remain_enabled():
|
|
source = '<html><body><script>document.execCommand("copy")</script></body></html>'
|
|
content = ttyd_patch.patch_html(source)
|
|
assert 'id="atlas-ttyd-clipboard"' in content
|
|
assert "navigator.clipboard.writeText(text)" in content
|
|
assert "class AtlasRecoveringWebSocket" in content
|
|
assert "window.location.reload()" in content
|
|
assert "event.stopImmediatePropagation()" in content
|
|
|
|
|
|
def test_ttyd_patch_fails_closed_on_upstream_drift():
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
ttyd_patch.patch_html("<html><body>changed</body></html>")
|