Files
雷电芽衣 1656a94811 sdks/python: mirror Elydora-Python-SDK 2.0.2
Rebuilt from the standalone 54e65eb tree (durable transactions, Kiro IDE
adapter, hook retry with a bounded chain lock, version 2.0.2) with the
Open-Source API trims re-applied: no token rotation, token issuance, or
agent integration updates, and the mirror's registration contract and
OpenCode plugin tests kept.

Claude-Session: https://claude.ai/code/session_01QoUjdc7HGKTsgj5GhG8HvF
2026-09-02 10:48:33 -04:00

259 lines
7.9 KiB
Python

from __future__ import annotations
import os
from pathlib import Path
import shutil
import subprocess # nosec B404
import sys
import pytest
from gemini_support import (
AGENT_ID,
VALID_PRIVATE_KEY,
prepare_fixture,
symlink_or_skip,
write_json,
write_text,
)
from elydora.plugins import _transaction, gemini_installation
from elydora.plugins.gemini_config import render_gemini_document
from elydora.plugins.gemini_contract import (
AUDIT_HOOK_NAME,
GUARD_HOOK_NAME,
build_gemini_group,
)
from elydora.plugins.gemini_io import read_gemini_document
RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py")
def prepare_installation(fixture: object) -> list[_transaction.FileChange]:
document = read_gemini_document()
paths = gemini_installation.preflight_gemini_installation(
fixture.config, document
)
rendered = render_gemini_document(
document,
None,
{
"BeforeTool": build_gemini_group(
paths.guard_path, GUARD_HOOK_NAME
),
"AfterTool": build_gemini_group(
paths.audit_path, AUDIT_HOOK_NAME
),
},
)
return gemini_installation.prepare_gemini_installation(
fixture.config, paths, rendered
)
def assert_no_transaction_files(root: Path) -> None:
assert not list(root.rglob("*.tmp"))
assert not list(root.rglob("*.rollback"))
def test_transaction_rolls_back_all_five_files(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
original = '{"theme":"GitHub"}\n'
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_settings=original
)
real_replace = _transaction._replace_physical
failed = False
def fail_settings_once(
source: str,
destination: str,
directory: _transaction.PinnedDirectory,
) -> None:
nonlocal failed
if os.path.abspath(destination) == str(fixture.config_path) and not failed:
failed = True
raise OSError("injected Gemini settings failure")
real_replace(source, destination, directory)
monkeypatch.setattr(_transaction, "_replace_physical", fail_settings_once)
with pytest.raises(OSError, match="injected Gemini settings failure"):
fixture.install()
assert fixture.source() == original
for name in RUNTIME_NAMES:
assert not (fixture.agent_dir / name).exists()
assert_no_transaction_files(fixture.home_dir)
def test_concurrent_settings_change_is_detected_before_runtime_commit(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_settings={"theme": "GitHub"}
)
changes = prepare_installation(fixture)
concurrent = '{"theme":"Atom One","hooks":{"Notification":[]}}\n'
write_text(fixture.config_path, concurrent)
with pytest.raises(OSError, match="changed during installation"):
gemini_installation.commit_gemini_installation(changes)
assert fixture.source() == concurrent
for name in RUNTIME_NAMES:
assert not (fixture.agent_dir / name).exists()
assert_no_transaction_files(fixture.home_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_settings_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.config_path.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_path, {
"agent_id": "another-agent", "agent_name": "gemini"
})
original = fixture.runtime_config_path.read_text(encoding="utf-8")
with pytest.raises(ValueError, match="identity does not match"):
fixture.install()
assert fixture.runtime_config_path.read_text(encoding="utf-8") == original
assert not fixture.config_path.exists()
@pytest.mark.parametrize("kind", ["configuration", "settings", "runtime"])
def test_linked_configuration_and_runtime_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 == "settings":
write_text(target, "{}\n")
fixture.config_path.parent.mkdir(parents=True)
symlink_or_skip(target, fixture.config_path)
else:
target.mkdir()
link = (
fixture.config_path.parent
if kind == "configuration"
else fixture.agent_dir.parent
)
if kind == "runtime" and link.exists():
shutil.rmtree(link)
link.parent.mkdir(parents=True, exist_ok=True)
symlink_or_skip(target, link, directory=True)
with pytest.raises(OSError, match="physical (file|directory)"):
fixture.install()
@pytest.mark.parametrize(
("field", "value", "pattern"),
[
("agent_name", "codex", "requires agent_name gemini"),
("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.config_path.exists()
assert not fixture.agent_dir.exists()
def test_status_surfaces_malformed_runtime_metadata(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.install()
write_text(fixture.runtime_config_path, "{ malformed")
with pytest.raises(ValueError, match="parse Elydora runtime config"):
fixture.plugin.status()
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.home_dir)
def test_cli_preflight_preserves_malformed_settings(tmp_path: Path) -> None:
home_dir = tmp_path / "cli-home"
settings_path = home_dir / ".gemini" / "settings.json"
key_path = tmp_path / "private.key"
write_text(settings_path, "{ malformed")
write_text(key_path, VALID_PRIVATE_KEY)
environment = {
**os.environ,
"HOME": str(home_dir),
"USERPROFILE": str(home_dir),
"GEMINI_CLI_HOME": str(home_dir),
}
result = subprocess.run( # nosec B603
[
sys.executable, "-m", "elydora.cli", "install",
"--agent", "gemini", "--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 "parse Gemini CLI user settings" in result.stderr
assert settings_path.read_text(encoding="utf-8") == "{ malformed"
assert not (home_dir / ".elydora").exists()