fix(cline): mirror transactional Python adapter

This commit is contained in:
ZacharyZhang-NY committed 2026-07-20 11:11:34 -04:00
1 parent 4ad139c60a
commit 052116081a
10 files changed
+1375 -572

No files matched your search

+226
View File
@@ -0,0 +1,226 @@
from __future__ import annotations
import base64
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import shutil
import subprocess # nosec B404
import sys
from threading import Thread
from typing import Any, Mapping
import pytest
from elydora.plugins import cline_contract as contract
from elydora.plugins.base import InstallConfig
from elydora.plugins.cline import ClinePlugin
AGENT_ID = "agent-1"
MISSING = object()
VALID_PRIVATE_KEY = base64.urlsafe_b64encode(bytes(range(32))).rstrip(b"=").decode()
@dataclass
class ClineFixture:
root_dir: Path
home_dir: Path
project_dir: Path
cline_dir: Path
hooks_dir: Path
agent_dir: Path
guard_path: Path
audit_path: Path
guard_wrapper: Path
audit_wrapper: Path
runtime_config: Path
private_key: Path
plugin: ClinePlugin
config: InstallConfig
def install(self) -> None:
self.plugin.install(self.config)
def environment(self) -> dict[str, str]:
return {
**os.environ,
"HOME": str(self.home_dir),
"USERPROFILE": str(self.home_dir),
"CLINE_DIR": str(self.cline_dir),
}
def managed_paths(self) -> tuple[Path, ...]:
return (
self.guard_path,
self.runtime_config,
self.private_key,
self.audit_path,
self.guard_wrapper,
self.audit_wrapper,
)
class ElydoraApiHandler(BaseHTTPRequestHandler):
agent_status: Any = "active"
requests: list[dict[str, Any]] = []
@classmethod
def reset(cls, status: Any) -> None:
cls.agent_status = status
cls.requests = []
def _respond(self, status: int, value: object) -> None:
body = json.dumps(value).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
type(self).requests.append(
{
"method": "GET",
"path": self.path,
"authorization": self.headers.get("Authorization"),
}
)
self._respond(200, {"agent": {"status": type(self).agent_status}})
def do_POST(self) -> None: # noqa: N802
length = int(self.headers.get("Content-Length", "0"))
raw = self.rfile.read(length)
type(self).requests.append(
{
"method": "POST",
"path": self.path,
"authorization": self.headers.get("Authorization"),
"raw": raw,
"json": json.loads(raw),
}
)
self._respond(201, {"operation": {"accepted": True}})
def log_message(self, _format: str, *_args: object) -> None:
return
def start_api_server(*, status: Any = "active") -> ThreadingHTTPServer:
ElydoraApiHandler.reset(status)
server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler)
Thread(target=server.serve_forever, daemon=True).start()
return server
def server_base_url(server: ThreadingHTTPServer) -> str:
return f"http://127.0.0.1:{server.server_port}"
def write_text(path: Path, source: str, mode: int = 0o600) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(source, encoding="utf-8", newline="")
os.chmod(path, mode)
def write_json(path: Path, value: object, mode: int = 0o600) -> None:
source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n"
write_text(path, source, mode)
def python_script(body: str) -> str:
return f"#!{sys.executable}\n{body}"
def prepare_fixture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
*,
existing_guard: object = MISSING,
existing_audit: object = MISSING,
base_url: str = "http://127.0.0.1:9",
) -> ClineFixture:
home_dir = tmp_path / "home with spaces and 'quote %CLINE%"
project_dir = tmp_path / "project with spaces"
cline_dir = tmp_path / "custom-cline-home"
hooks_dir = cline_dir / "hooks"
agent_dir = home_dir / ".elydora" / AGENT_ID
project_dir.mkdir(parents=True)
monkeypatch.setenv("HOME", str(home_dir))
monkeypatch.setenv("USERPROFILE", str(home_dir))
monkeypatch.setenv("CLINE_DIR", str(cline_dir))
monkeypatch.chdir(project_dir)
guard_wrapper = hooks_dir / contract.GUARD_FILE_NAME
audit_wrapper = hooks_dir / contract.AUDIT_FILE_NAME
if existing_guard is not MISSING:
write_text(guard_wrapper, str(existing_guard), 0o700)
if existing_audit is not MISSING:
write_text(audit_wrapper, str(existing_audit), 0o700)
guard_path = agent_dir / contract.GUARD_SCRIPT
audit_path = agent_dir / contract.AUDIT_SCRIPT
return ClineFixture(
root_dir=tmp_path,
home_dir=home_dir,
project_dir=project_dir,
cline_dir=cline_dir,
hooks_dir=hooks_dir,
agent_dir=agent_dir,
guard_path=guard_path,
audit_path=audit_path,
guard_wrapper=guard_wrapper,
audit_wrapper=audit_wrapper,
runtime_config=agent_dir / "config.json",
private_key=agent_dir / "private.key",
plugin=ClinePlugin(),
config={
"org_id": "org-1",
"agent_id": AGENT_ID,
"agent_name": "cline",
"private_key": VALID_PRIVATE_KEY,
"kid": "key-1",
"token": "token-1",
"base_url": base_url,
"guard_script_path": str(guard_path),
},
)
def run_wrapper(
wrapper: Path,
fixture: ClineFixture,
payload: bytes,
) -> subprocess.CompletedProcess[bytes]:
node = shutil.which("node")
assert node is not None
return subprocess.run( # nosec B603
[node, str(wrapper)],
cwd=fixture.project_dir,
env=fixture.environment(),
input=payload,
capture_output=True,
check=False,
timeout=10,
)
def assert_no_transaction_files(root: Path) -> None:
assert not list(root.rglob("*.tmp"))
assert not list(root.rglob("*.rollback"))
def snapshot_installation(fixture: ClineFixture) -> Mapping[Path, bytes]:
return {path: path.read_bytes() for path in fixture.managed_paths()}
def assert_snapshot(snapshot: Mapping[Path, bytes]) -> None:
for path, source in snapshot.items():
assert path.read_bytes() == source
def symlink_or_skip(target: Path, link: Path, *, directory: bool = False) -> None:
try:
link.symlink_to(target, target_is_directory=directory)
except OSError as error:
pytest.skip(f"Symbolic links unavailable: {error}")
+81
View File
@@ -0,0 +1,81 @@
from __future__ import annotations
import json
import os
from pathlib import Path
import shutil
import subprocess # nosec B404
import pytest
from cline_support import prepare_fixture
CLINE_CORE_ENTRY = os.environ.get("ELYDORA_CLINE_CORE_ENTRY")
CLINE_ENTRY = os.environ.get("ELYDORA_CLINE_ENTRY")
@pytest.mark.skipif(
not (CLINE_CORE_ENTRY and CLINE_ENTRY),
reason=(
"set ELYDORA_CLINE_CORE_ENTRY and ELYDORA_CLINE_ENTRY "
"to official Cline files"
),
)
def test_official_cline_loader_discovers_both_managed_hooks(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
node = shutil.which("node")
assert node is not None
assert CLINE_CORE_ENTRY is not None
assert CLINE_ENTRY is not None
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
version = subprocess.run( # nosec B603
[node, CLINE_ENTRY, "--version"],
cwd=fixture.project_dir,
env=fixture.environment(),
capture_output=True,
check=False,
text=True,
timeout=10,
)
assert version.returncode == 0, version.stderr
assert version.stdout.strip() == "3.0.46"
source = """
import { pathToFileURL } from 'node:url';
const { listHookConfigFiles } = await import(
pathToFileURL(process.env.ELYDORA_CLINE_CORE_ENTRY).href
);
console.log(JSON.stringify(listHookConfigFiles(process.env.ELYDORA_WORKSPACE)));
"""
environment = {
**fixture.environment(),
"ELYDORA_CLINE_CORE_ENTRY": CLINE_CORE_ENTRY,
"ELYDORA_WORKSPACE": str(fixture.project_dir),
}
loaded = subprocess.run( # nosec B603
[node, "--input-type=module", "--eval", source],
cwd=fixture.project_dir,
env=environment,
capture_output=True,
check=False,
text=True,
timeout=10,
)
assert loaded.returncode == 0, loaded.stderr
assert json.loads(loaded.stdout) == [
{
"fileName": "PostToolUse",
"hookEventName": "tool_result",
"path": str(fixture.audit_wrapper),
},
{
"fileName": "PreToolUse",
"hookEventName": "tool_call",
"path": str(fixture.guard_wrapper),
},
]
+285 -345
View File
@@ -1,143 +1,66 @@
from __future__ import annotations
import base64
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
from queue import Queue
import shutil
import subprocess # nosec B404
import sys
from threading import Thread
from typing import Any, Optional
import pytest
from cline_support import (
AGENT_ID,
ElydoraApiHandler,
MISSING,
assert_no_transaction_files,
prepare_fixture,
python_script,
run_wrapper,
server_base_url,
start_api_server,
write_text,
)
from elydora import cli
from elydora.plugins import cline
from elydora.plugins import cline_contract, cline_io
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS
AGENT_ID = "agent-1"
MISSING = object()
def encoded(value: object) -> bytes:
return json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode()
@dataclass(frozen=True)
class ClineFixture:
plugin: cline.ClinePlugin
config: InstallConfig
home_dir: Path
workspace_dir: Path
cline_dir: Path
hooks_dir: Path
agent_dir: Path
guard_path: Path
audit_path: Path
guard_wrapper: Path
audit_wrapper: Path
runtime_config: Path
def _python_script(body: str) -> str:
return f"#!{sys.executable}\n{body}"
def prepare_fixture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
*,
auto_install: bool = True,
create_guard: bool = True,
existing_audit: Any = MISSING,
existing_guard: Any = MISSING,
guard_source: Optional[str] = None,
) -> ClineFixture:
home_dir = tmp_path / "home with spaces"
workspace_dir = tmp_path / "workspace"
cline_dir = tmp_path / "custom-cline-home"
hooks_dir = cline_dir / "hooks"
agent_dir = home_dir / ".elydora" / AGENT_ID
guard_path = agent_dir / "guard.py"
audit_path = agent_dir / "hook.py"
guard_wrapper = hooks_dir / "PreToolUse.mjs"
audit_wrapper = hooks_dir / "PostToolUse.mjs"
workspace_dir.mkdir(parents=True)
agent_dir.mkdir(parents=True)
if create_guard:
guard_path.write_text(
guard_source
or _python_script(
"import sys\n"
"sys.stdin.read()\n"
"sys.stderr.write('Agent is frozen by Elydora.\\n')\n"
"raise SystemExit(2)\n"
),
encoding="utf-8",
)
if existing_guard is not MISSING:
hooks_dir.mkdir(parents=True, exist_ok=True)
guard_wrapper.write_text(str(existing_guard), encoding="utf-8")
if existing_audit is not MISSING:
hooks_dir.mkdir(parents=True, exist_ok=True)
audit_wrapper.write_text(str(existing_audit), encoding="utf-8")
monkeypatch.setenv("HOME", str(home_dir))
monkeypatch.setenv("USERPROFILE", str(home_dir))
monkeypatch.setenv("CLINE_DIR", str(cline_dir))
private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode()
config: InstallConfig = {
"agent_id": AGENT_ID,
"agent_name": "cline",
"org_id": "org-1",
"private_key": private_key,
"kid": "kid-1",
"base_url": "https://api.elydora.test",
"guard_script_path": str(guard_path),
def official_payload(event: str) -> dict[str, object]:
nested = {
"id": "tool-call-1",
"name": "read_file",
"input": {"path": "README.md", "nested": {"preserve": True}},
}
fixture = ClineFixture(
plugin=cline.ClinePlugin(),
config=config,
home_dir=home_dir,
workspace_dir=workspace_dir,
cline_dir=cline_dir,
hooks_dir=hooks_dir,
agent_dir=agent_dir,
guard_path=guard_path,
audit_path=audit_path,
guard_wrapper=guard_wrapper,
audit_wrapper=audit_wrapper,
runtime_config=agent_dir / "config.json",
)
if auto_install:
fixture.plugin.install(config)
return fixture
def run_wrapper(
wrapper: Path,
fixture: ClineFixture,
payload: dict[str, Any],
) -> subprocess.CompletedProcess[str]:
node = shutil.which("node")
assert node is not None
return subprocess.run( # nosec B603
[node, str(wrapper)],
cwd=fixture.workspace_dir,
env={
**os.environ,
"HOME": str(fixture.home_dir),
"USERPROFILE": str(fixture.home_dir),
},
input=json.dumps(payload),
text=True,
capture_output=True,
check=False,
timeout=10,
)
value: dict[str, object] = {
"clineVersion": "3.0.46",
"hookName": event,
"timestamp": "2026-07-19T12:00:00.000Z",
"taskId": "task-1",
"workspaceRoots": ["/workspace"],
"userId": "user-1",
"futureField": {"survives": ["exactly", 2]},
}
if event == "tool_call":
value["tool_call"] = nested
value["preToolUse"] = {
"toolName": nested["name"],
"parameters": nested["input"],
}
else:
value["tool_result"] = {
**nested,
"output": "ok",
"durationMs": 12,
}
value["postToolUse"] = {
"toolName": nested["name"],
"parameters": nested["input"],
"result": "ok",
"success": True,
"executionTimeMs": 12,
}
return value
def test_cline_is_registered_in_the_sdk_and_cli() -> None:
@@ -147,35 +70,46 @@ def test_cline_is_registered_in_the_sdk_and_cli() -> None:
"config_path": "~/.cline/hooks/PreToolUse.mjs",
}
assert cli.PLUGIN_MAP["cline"] is cline.ClinePlugin
assert cline.ClinePlugin.manages_guard_runtime is True
def test_install_writes_only_native_global_hooks_and_is_idempotent(
def test_install_commits_all_six_files_and_is_idempotent(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
guard_source = fixture.guard_wrapper.read_text(encoding="utf-8")
audit_source = fixture.audit_wrapper.read_text(encoding="utf-8")
fixture.install()
before = {path: path.read_bytes() for path in fixture.managed_paths()}
fixture.plugin.install(fixture.config)
fixture.install()
assert fixture.guard_wrapper.read_text(encoding="utf-8") == guard_source
assert fixture.audit_wrapper.read_text(encoding="utf-8") == audit_source
assert guard_source.startswith("#!/usr/bin/env node\n// @elydora-cline-hook ")
assert list(fixture.hooks_dir.glob("*.tmp")) == []
assert {path: path.read_bytes() for path in fixture.managed_paths()} == before
assert fixture.guard_wrapper.read_text(encoding="utf-8").startswith(
"#!/usr/bin/env node\n// @elydora-cline-hook "
)
assert json.loads(fixture.runtime_config.read_text(encoding="utf-8")) == {
"org_id": "org-1",
"agent_id": AGENT_ID,
"kid": "key-1",
"base_url": "http://127.0.0.1:9",
"agent_name": "cline",
"token": "token-1",
}
assert not (fixture.home_dir / "Documents" / "Cline" / "Hooks").exists()
assert not (fixture.workspace_dir / ".cline" / "hooks").exists()
assert not (fixture.workspace_dir / ".clinerules" / "hooks").exists()
assert not (fixture.project_dir / ".cline" / "hooks").exists()
assert not (fixture.project_dir / ".clinerules" / "hooks").exists()
assert_no_transaction_files(fixture.home_dir)
assert_no_transaction_files(fixture.cline_dir)
def test_install_uses_official_default_without_cline_dir(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False)
fixture = prepare_fixture(monkeypatch, tmp_path)
monkeypatch.delenv("CLINE_DIR")
fixture.plugin.install(fixture.config)
fixture.install()
default_hooks = fixture.home_dir / ".cline" / "hooks"
assert (default_hooks / "PreToolUse.mjs").is_file()
@@ -183,184 +117,234 @@ def test_install_uses_official_default_without_cline_dir(
assert not fixture.guard_wrapper.exists()
def test_wrappers_translate_freezes_and_forward_payload_byte_for_byte(
def test_wrappers_preserve_bytes_and_emit_pure_json_control(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
capture_path = tmp_path / "captured-event.json"
fixture.audit_path.write_text(
_python_script(
fixture.install()
guard_capture = tmp_path / "guard-input.json"
audit_capture = tmp_path / "audit-input.json"
write_text(
fixture.guard_path,
python_script(
"from pathlib import Path\n"
"import os, sys\n"
f"Path({str(capture_path)!r}).write_text("
"json.dumps({'cwd': os.getcwd(), 'input': sys.stdin.read()}), "
"encoding='utf-8')\n"
).replace("import os, sys", "import json, os, sys"),
encoding="utf-8",
"import sys\n"
f"Path({str(guard_capture)!r}).write_bytes(sys.stdin.buffer.read())\n"
"sys.stderr.write('Agent is frozen by Elydora.\\n')\n"
"raise SystemExit(2)\n"
),
0o700,
)
pre_payload = {
"clineVersion": "3.0.46",
"hookName": "tool_call",
"taskId": "task-1",
"tool_call": {"name": "read_file", "input": {"path": "README.md"}},
}
guard = run_wrapper(fixture.guard_wrapper, fixture, pre_payload)
write_text(
fixture.audit_path,
python_script(
"from pathlib import Path\n"
"import sys\n"
f"Path({str(audit_capture)!r}).write_bytes(sys.stdin.buffer.read())\n"
),
0o700,
)
pre_raw = encoded(official_payload("tool_call"))
post_raw = encoded(official_payload("tool_result"))
guard = run_wrapper(fixture.guard_wrapper, fixture, pre_raw)
audit = run_wrapper(fixture.audit_wrapper, fixture, post_raw)
assert guard.returncode == 0
assert "Agent is frozen by Elydora" in guard.stderr
marker, control = guard.stdout.strip().split("\t", 1)
assert marker == "HOOK_CONTROL"
assert json.loads(control) == {
assert json.loads(guard.stdout) == {
"cancel": True,
"errorMessage": "Agent is frozen by Elydora.",
}
post_payload = {
"clineVersion": "3.0.46",
"hookName": "tool_result",
"taskId": "task-1",
"tool_result": {"name": "read_file", "input": {"path": "README.md"}},
}
audit = run_wrapper(fixture.audit_wrapper, fixture, post_payload)
assert b"HOOK_CONTROL" not in guard.stdout
assert guard_capture.read_bytes() == pre_raw
assert audit.returncode == 0
captured = json.loads(capture_path.read_text(encoding="utf-8"))
assert captured == {"cwd": str(fixture.workspace_dir), "input": json.dumps(post_payload)}
assert audit.stdout == b""
assert audit_capture.read_bytes() == post_raw
@pytest.mark.parametrize("status", ["frozen", "revoked"])
def test_generated_guard_translates_blocking_status_to_cline_control(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
status: str,
) -> None:
server = start_api_server(status=status)
try:
fixture = prepare_fixture(
monkeypatch, tmp_path, base_url=server_base_url(server)
)
fixture.install()
result = run_wrapper(
fixture.guard_wrapper,
fixture,
encoded(official_payload("tool_call")),
)
finally:
server.shutdown()
server.server_close()
assert result.returncode == 0
assert json.loads(result.stdout)["cancel"] is True
assert status.encode() in result.stderr.lower()
def test_wrappers_keep_passes_quiet_and_surface_runtime_failures(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
passing = prepare_fixture(
monkeypatch,
tmp_path / "passing",
guard_source=_python_script("import sys\nsys.stdin.read()\n"),
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
write_text(
fixture.guard_path,
python_script("import sys\nsys.stdin.buffer.read()\n"),
0o700,
)
result = run_wrapper(passing.guard_wrapper, passing, {})
assert result.returncode == 0
assert result.stdout == ""
passing = run_wrapper(fixture.guard_wrapper, fixture, b"{}")
assert passing.returncode == 0
assert passing.stdout == b""
failing = prepare_fixture(monkeypatch, tmp_path / "failing")
failing.audit_path.write_text(
_python_script(
write_text(
fixture.audit_path,
python_script(
"import sys\n"
"sys.stderr.write('audit failed\\n')\n"
"raise SystemExit(7)\n"
),
encoding="utf-8",
0o700,
)
result = run_wrapper(failing.audit_wrapper, failing, {})
assert result.returncode == 1
assert "audit failed" in result.stderr
assert "exited with code 7" in result.stderr
failing = run_wrapper(fixture.audit_wrapper, fixture, b"{}")
assert failing.returncode == 1
assert b"audit failed" in failing.stderr
assert b"exited with code 7" in failing.stderr
def test_wrapper_rejects_a_non_absolute_runtime_shebang(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
guard_source="#!python.exe\nimport sys\nsys.stdin.read()\n",
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
write_text(
fixture.guard_path,
"#!python.exe\nimport sys\nsys.stdin.buffer.read()\n",
0o700,
)
result = run_wrapper(fixture.guard_wrapper, fixture, {})
result = run_wrapper(fixture.guard_wrapper, fixture, b"{}")
assert result.returncode == 1
assert result.stdout == ""
assert "runtime shebang executable must be absolute" in result.stderr
assert result.stdout == b""
assert b"runtime shebang executable must be absolute" in result.stderr
class _OperationHandler(BaseHTTPRequestHandler):
operations: Queue[dict[str, Any]]
def do_POST(self) -> None: # noqa: N802
length = int(self.headers["Content-Length"])
body = json.loads(self.rfile.read(length).decode("utf-8"))
self.operations.put(body)
self.send_response(201)
self.end_headers()
self.wfile.write(b"{}")
def log_message(self, format: str, *args: Any) -> None:
return
def test_audit_runtime_maps_official_nested_fields_into_the_operation(
def test_audit_submits_the_complete_native_event(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
operations: Queue[dict[str, Any]] = Queue()
_OperationHandler.operations = operations
server = ThreadingHTTPServer(("127.0.0.1", 0), _OperationHandler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False)
fixture.config["base_url"] = f"http://127.0.0.1:{server.server_port}"
server = start_api_server()
payload = official_payload("tool_result")
try:
fixture.plugin.install(fixture.config)
payload = {
"hookName": "tool_result",
"taskId": "task-1",
"tool_result": {
"name": "read_file",
"input": {"path": "README.md"},
"output": "ok",
},
}
result = run_wrapper(fixture.audit_wrapper, fixture, payload)
assert result.returncode == 0
operation = operations.get(timeout=3)
assert operation["payload"] == {
"tool_name": "read_file",
"tool_input": {"path": "README.md"},
"session_id": "task-1",
}
assert operation["action"] == {"tool": "read_file"}
assert operation["subject"] == {"session_id": "task-1"}
fixture = prepare_fixture(
monkeypatch, tmp_path, base_url=server_base_url(server)
)
fixture.install()
result = run_wrapper(fixture.audit_wrapper, fixture, encoded(payload))
finally:
server.shutdown()
server.server_close()
thread.join(timeout=3)
def test_status_requires_intact_hooks_and_runtimes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
assert fixture.plugin.status() == {
"installed": True,
"agent": "cline",
"details": f"Config: {fixture.hooks_dir}",
}
fixture.guard_path.unlink()
assert fixture.plugin.status()["installed"] is False
fixture.guard_path.write_text(_python_script("pass\n"), encoding="utf-8")
fixture.audit_wrapper.unlink()
assert fixture.plugin.status() == {
"installed": False,
"agent": "cline",
"details": "Not installed",
}
def test_status_surfaces_corrupt_hooks_and_runtime_metadata(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.guard_wrapper.write_text(
fixture.guard_wrapper.read_text(encoding="utf-8") + "\n// tampered\n",
encoding="utf-8",
assert result.returncode == 0
request = next(
item for item in ElydoraApiHandler.requests if item["method"] == "POST"
)
with pytest.raises(ValueError, match="managed template"):
fixture.plugin.status()
operation = request["json"]
assert operation["payload"] == payload
assert operation["action"] == {"tool": "read_file"}
assert operation["subject"] == {"session_id": "task-1"}
assert request["authorization"] == "Bearer token-1"
fixture.plugin.install(fixture.config)
fixture.runtime_config.write_text("{ malformed", encoding="utf-8")
with pytest.raises(ValueError, match="parse Elydora runtime config"):
@pytest.mark.parametrize(
("target", "expected"),
[
("guard", None),
("audit", None),
("config", "parse Elydora runtime config"),
("key", "private key.*canonical 32-byte"),
("wrapper", "managed template"),
],
)
def test_status_requires_exact_physical_runtime_identity(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
target: str,
expected: str | None,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
assert fixture.plugin.status()["installed"] is True
if target == "guard":
write_text(fixture.guard_path, "tampered\n", 0o700)
elif target == "audit":
write_text(fixture.audit_path, "tampered\n", 0o700)
elif target == "config":
write_text(fixture.runtime_config, "{ malformed")
elif target == "key":
write_text(fixture.private_key, "invalid")
else:
source = fixture.guard_wrapper.read_text(encoding="utf-8")
write_text(fixture.guard_wrapper, source + "// tampered\n", 0o700)
if expected is None:
assert fixture.plugin.status()["installed"] is False
else:
with pytest.raises(ValueError, match=expected):
fixture.plugin.status()
@pytest.mark.parametrize(
("source", "pattern"),
[
('{"agent_name":"cline","agent_name":"cline"}', "duplicate field"),
(
json.dumps(
{
"org_id": "org-1",
"agent_id": "other-agent",
"kid": "key-1",
"base_url": "https://api.elydora.com",
"agent_name": "cline",
}
),
"identity does not match",
),
(
json.dumps(
{
"org_id": "org-1",
"agent_id": AGENT_ID,
"kid": "key-1",
"base_url": "https://api.elydora.com",
"agent_name": "cline",
"hidden": True,
}
),
"unsupported field",
),
],
)
def test_status_surfaces_invalid_runtime_metadata(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
source: str,
pattern: str,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
write_text(fixture.runtime_config, source)
with pytest.raises(ValueError, match=pattern):
fixture.plugin.status()
@@ -370,25 +354,21 @@ def test_install_rejects_user_filename_collisions_before_writes(
tmp_path: Path,
collision: str,
) -> None:
existing_guard = "// user PreToolUse hook\n" if collision == "guard" else MISSING
existing_audit = "// user PostToolUse hook\n" if collision == "audit" else MISSING
guard = "// user PreToolUse hook\n" if collision == "guard" else MISSING
audit = "// user PostToolUse hook\n" if collision == "audit" else MISSING
fixture = prepare_fixture(
monkeypatch,
tmp_path,
auto_install=False,
existing_guard=existing_guard,
existing_audit=existing_audit,
existing_guard=guard,
existing_audit=audit,
)
with pytest.raises(ValueError, match="owned by another integration"):
fixture.plugin.install(fixture.config)
assert fixture.audit_path.exists() is False
assert fixture.runtime_config.exists() is False
if collision == "guard":
assert fixture.guard_wrapper.read_text(encoding="utf-8") == existing_guard
assert fixture.audit_wrapper.exists() is False
else:
assert fixture.audit_wrapper.read_text(encoding="utf-8") == existing_audit
assert fixture.guard_wrapper.exists() is False
fixture.install()
assert not fixture.agent_dir.exists()
path = fixture.guard_wrapper if collision == "guard" else fixture.audit_wrapper
assert path.read_text(encoding="utf-8").startswith("// user")
def test_install_preserves_corrupt_metadata_for_recovery(
@@ -396,88 +376,48 @@ def test_install_preserves_corrupt_metadata_for_recovery(
tmp_path: Path,
) -> None:
corrupt = "#!/usr/bin/env node\n// @elydora-cline-hook invalid\n"
fixture = prepare_fixture(
monkeypatch,
tmp_path,
auto_install=False,
existing_guard=corrupt,
)
fixture = prepare_fixture(monkeypatch, tmp_path, existing_guard=corrupt)
with pytest.raises(ValueError, match="parse Elydora Cline hook metadata"):
fixture.plugin.install(fixture.config)
fixture.install()
assert fixture.guard_wrapper.read_text(encoding="utf-8") == corrupt
assert fixture.audit_wrapper.exists() is False
assert fixture.audit_path.exists() is False
assert not fixture.agent_dir.exists()
def test_install_rejects_missing_guard_before_creating_files(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
auto_install=False,
create_guard=False,
)
with pytest.raises(FileNotFoundError, match="guard runtime is missing"):
fixture.plugin.install(fixture.config)
assert fixture.guard_wrapper.exists() is False
assert fixture.audit_wrapper.exists() is False
assert fixture.audit_path.exists() is False
assert fixture.runtime_config.exists() is False
def test_uninstall_removes_exact_ownership_and_preserves_other_hooks(
def test_uninstall_removes_exact_ownership_and_preserves_adjacent_hooks(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
user_hook = fixture.hooks_dir / "PreToolUse.py"
user_hook.write_text("# user hook\n", encoding="utf-8")
write_text(user_hook, "# user hook\n")
fixture.plugin.uninstall("agent-10")
assert fixture.guard_wrapper.is_file()
assert fixture.audit_wrapper.is_file()
fixture.plugin.uninstall(AGENT_ID)
assert fixture.guard_wrapper.exists() is False
assert fixture.audit_wrapper.exists() is False
assert not fixture.guard_wrapper.exists()
assert not fixture.audit_wrapper.exists()
assert user_hook.read_text(encoding="utf-8") == "# user hook\n"
def test_hook_pair_restores_the_first_hook_when_the_second_commit_fails(
def test_runtime_failures_remain_observable_and_fail_open(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
original_guard = fixture.guard_wrapper.read_text(encoding="utf-8")
original_audit = fixture.audit_wrapper.read_text(encoding="utf-8")
guard_state = cline_io.read_hook_file(str(fixture.guard_wrapper))
audit_state = cline_io.read_hook_file(str(fixture.audit_wrapper))
replacement_guard = cline_contract.build_wrapper(
cline_contract.build_metadata("guard", "agent-2", str(fixture.guard_path))
fixture.install()
guard = run_wrapper(
fixture.guard_wrapper,
fixture,
encoded(official_payload("tool_call")),
)
replacement_audit = cline_contract.build_wrapper(
cline_contract.build_metadata("audit", "agent-2", str(fixture.audit_path))
)
real_replace = os.replace
failure_injected = False
audit = run_wrapper(fixture.audit_wrapper, fixture, b"{ malformed")
def fail_audit_commit(source: Any, destination: Any) -> None:
nonlocal failure_injected
if not failure_injected and Path(destination) == fixture.audit_wrapper:
failure_injected = True
raise OSError("simulated audit commit failure")
real_replace(source, destination)
monkeypatch.setattr(os, "replace", fail_audit_commit)
with pytest.raises(OSError, match="Write Cline hook pair"):
cline_io.write_hook_pair(
cline_io.PendingWrite(guard_state, replacement_guard),
cline_io.PendingWrite(audit_state, replacement_audit),
)
assert failure_injected is True
assert fixture.guard_wrapper.read_text(encoding="utf-8") == original_guard
assert fixture.audit_wrapper.read_text(encoding="utf-8") == original_audit
assert list(fixture.hooks_dir.glob("*.tmp")) == []
assert guard.returncode == 0
assert b"Failed to resolve agent status" in guard.stderr
assert audit.returncode == 0
assert b"invalid JSON" in (fixture.agent_dir / "error.log").read_bytes()
+304
View File
@@ -0,0 +1,304 @@
from __future__ import annotations
import os
from pathlib import Path
import subprocess # nosec B404
import sys
import pytest
from cline_support import (
AGENT_ID,
VALID_PRIVATE_KEY,
assert_no_transaction_files,
assert_snapshot,
prepare_fixture,
snapshot_installation,
symlink_or_skip,
write_json,
write_text,
)
from elydora.plugins import _transaction, cline_installation
from elydora.plugins.cline_contract import resolve_hook_files
from elydora.plugins.cline_io import read_hook_file
def prepare_installation(fixture: object) -> list[_transaction.FileChange]:
paths = resolve_hook_files()
guard = read_hook_file(paths.guard_path)
audit = read_hook_file(paths.audit_path)
return cline_installation.prepare_cline_installation(
fixture.config, guard, audit
)
def test_transaction_rolls_back_all_six_files(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
real_replace = _transaction.os.replace
failed = False
def fail_audit_once(source: str, destination: str) -> None:
nonlocal failed
if os.path.abspath(destination) == str(fixture.audit_wrapper) and not failed:
failed = True
raise OSError("injected Cline audit hook failure")
real_replace(source, destination)
monkeypatch.setattr(_transaction.os, "replace", fail_audit_once)
with pytest.raises(OSError, match="injected Cline audit hook failure"):
fixture.install()
assert failed is True
assert all(not path.exists() for path in fixture.managed_paths())
assert_no_transaction_files(fixture.home_dir)
assert_no_transaction_files(fixture.cline_dir)
def test_concurrent_hook_replacement_is_detected_before_final_commit(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
changes = prepare_installation(fixture)
concurrent = "// concurrently replaced\n"
real_replace = _transaction.os.replace
injected = False
def inject_replacement(source: str, destination: str) -> None:
nonlocal injected
if not injected:
injected = True
write_text(fixture.audit_wrapper, concurrent, 0o700)
real_replace(source, destination)
monkeypatch.setattr(_transaction.os, "replace", inject_replacement)
with pytest.raises(OSError, match="changed during installation"):
cline_installation.commit_cline_installation(changes)
assert injected is True
assert fixture.audit_wrapper.read_text(encoding="utf-8") == concurrent
for path in fixture.managed_paths()[:-1]:
assert not path.exists()
assert_no_transaction_files(fixture.home_dir)
assert_no_transaction_files(fixture.cline_dir)
def test_stale_hook_snapshot_is_rejected_before_staging(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
paths = resolve_hook_files()
guard = read_hook_file(paths.guard_path)
audit = read_hook_file(paths.audit_path)
concurrent = "// stale snapshot replacement\n"
write_text(fixture.audit_wrapper, concurrent, 0o700)
with pytest.raises(OSError, match="changed before staging"):
cline_installation.prepare_cline_installation(
fixture.config, guard, audit
)
assert fixture.audit_wrapper.read_text(encoding="utf-8") == concurrent
assert not fixture.agent_dir.exists()
assert_no_transaction_files(fixture.root_dir)
@pytest.mark.parametrize(
"runtime_name",
[
"private.key",
"guard.py",
"hook.py",
"chain-state.json",
"status-cache.json",
"error.log",
],
)
def test_orphan_runtime_artifacts_fail_before_hook_writes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
runtime_name: str,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
artifact = fixture.agent_dir / runtime_name
write_text(artifact, "orphan\n")
with pytest.raises(ValueError, match="identity cannot be verified"):
fixture.install()
assert artifact.read_text(encoding="utf-8") == "orphan\n"
assert not fixture.guard_wrapper.exists()
assert not fixture.audit_wrapper.exists()
def test_mismatched_runtime_identity_fails_before_writes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
write_json(
fixture.runtime_config,
{"agent_id": "another-agent", "agent_name": "cline"},
)
original = fixture.runtime_config.read_text(encoding="utf-8")
with pytest.raises(ValueError, match="identity does not match"):
fixture.install()
assert fixture.runtime_config.read_text(encoding="utf-8") == original
assert not fixture.guard_wrapper.exists()
@pytest.mark.parametrize(
"kind", ["configuration", "hooks", "runtime", "hook"]
)
def test_linked_configuration_runtime_and_hook_paths_are_rejected(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
kind: str,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
target = tmp_path / f"{kind}-target"
if kind == "configuration":
target.mkdir()
symlink_or_skip(target, fixture.cline_dir, directory=True)
elif kind == "hooks":
target.mkdir()
fixture.cline_dir.mkdir()
symlink_or_skip(target, fixture.hooks_dir, directory=True)
elif kind == "runtime":
target.mkdir()
fixture.home_dir.mkdir()
symlink_or_skip(
target, fixture.home_dir / ".elydora", directory=True
)
else:
fixture.install()
write_text(target, "external hook\n")
fixture.guard_wrapper.unlink()
symlink_or_skip(target, fixture.guard_wrapper)
with pytest.raises(OSError, match="physical (file|directory)"):
fixture.install()
@pytest.mark.parametrize(
("field", "value", "pattern"),
[
("agent_name", "codex", "requires agent_name cline"),
("agent_id", "../escape", "Invalid agent ID"),
("private_key", "invalid", "canonical 32-byte"),
("token", "", "non-empty string"),
("base_url", "https://api.elydora.com?q=1", "query parameters"),
("guard_script_path", "outside", "managed agent directory"),
],
)
def test_install_validates_runtime_inputs_before_writes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
field: str,
value: str,
pattern: str,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.config[field] = value
with pytest.raises(ValueError, match=pattern):
fixture.install()
assert not fixture.agent_dir.exists()
assert not fixture.hooks_dir.exists()
def test_uninstall_restores_both_hooks_when_second_removal_fails(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
before = snapshot_installation(fixture)
paths = resolve_hook_files()
changes = cline_installation.prepare_cline_uninstall(
(
read_hook_file(paths.guard_path),
read_hook_file(paths.audit_path),
),
AGENT_ID,
)
real_replace = _transaction.os.replace
failed = False
def fail_audit_removal(source: str, destination: str) -> None:
nonlocal failed
if os.path.abspath(source) == str(fixture.audit_wrapper) and not failed:
failed = True
raise OSError("injected Cline uninstall failure")
real_replace(source, destination)
monkeypatch.setattr(_transaction.os, "replace", fail_audit_removal)
with pytest.raises(OSError, match="injected Cline uninstall failure"):
cline_installation.commit_cline_uninstall(changes)
assert failed is True
assert_snapshot(before)
assert_no_transaction_files(fixture.root_dir)
def test_installation_leaves_no_transaction_artifacts(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
assert_no_transaction_files(fixture.root_dir)
def test_cli_preflight_preserves_a_hook_collision(tmp_path: Path) -> None:
home_dir = tmp_path / "cli-home"
hook_path = home_dir / ".cline" / "hooks" / "PreToolUse.mjs"
key_path = tmp_path / "private.key"
write_text(hook_path, "// user hook\n", 0o700)
write_text(key_path, VALID_PRIVATE_KEY)
environment = {
key: value
for key, value in {
**os.environ,
"HOME": str(home_dir),
"USERPROFILE": str(home_dir),
}.items()
if key != "CLINE_DIR"
}
result = subprocess.run( # nosec B603
[
sys.executable,
"-m",
"elydora.cli",
"install",
"--agent",
"cline",
"--org_id",
"org-1",
"--agent_id",
AGENT_ID,
"--private_key_file",
str(key_path),
"--kid",
"key-1",
"--base_url",
"https://api.elydora.com",
],
capture_output=True,
check=False,
cwd=Path(__file__).parents[1],
env=environment,
text=True,
)
assert result.returncode == 1
assert "owned by another integration" in result.stderr
assert hook_path.read_text(encoding="utf-8") == "// user hook\n"
assert not (home_dir / ".elydora").exists()