diff --git a/README.md b/README.md index dedf24c..f2cf66d 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,18 @@ Each plugin ships both manifest flavors so it works everywhere today: The skill content is generated from the canonical source in the Delx Protocol repository — edit it there, then run `tools/sync-from-canonical.sh`. +## Tests + +Local unittest suite (no GitHub Actions — run on your machine): + +```bash +python3 -m unittest discover -s tests -t . -v +``` + +or `bash tests/run.sh`. Covers Agent Plugins manifests, MCP URLs, skill/tool +names, secret scanning, Continuity Capsule fields, and guardian-hook opt-in +behavior. + ## License Apache-2.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..9363e7b --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,2 @@ +# Plugin invariant tests. Run from the repo root: +# python3 -m unittest discover -s tests -t . -v diff --git a/tests/fixtures/continuity-capsule-v1.json b/tests/fixtures/continuity-capsule-v1.json new file mode 100644 index 0000000..e1bb58a --- /dev/null +++ b/tests/fixtures/continuity-capsule-v1.json @@ -0,0 +1,27 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://api.delx.ai/schemas/continuity-capsule-v1.json", + "title": "Delx Continuity Capsule v1", + "type": "object", + "additionalProperties": false, + "required": ["version"], + "properties": { + "version": { "type": "string", "enum": ["1", "1.0", "v1"] }, + "goal": { "type": "string", "maxLength": 4000 }, + "done": { "type": "string", "maxLength": 4000 }, + "next": { "type": "string", "maxLength": 4000 }, + "blockers": { "type": "string", "maxLength": 4000 }, + "do_not": { "type": "string", "maxLength": 4000 }, + "refuted": { "type": "string", "maxLength": 4000 }, + "receipts": { + "type": "array", + "maxItems": 20, + "items": { "type": "string", "maxLength": 500 } + }, + "written_by": { "type": "string", "maxLength": 200 }, + "written_at": { "type": "string" }, + "ttl_days": { "type": ["string", "integer"] }, + "handoff_to": { "type": "string", "maxLength": 200 } + }, + "description": "Snapshot of the Continuity Capsule v1 schema for offline tests. Live copy: https://api.delx.ai/schemas/continuity-capsule-v1.json" +} diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..82d3e86 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,107 @@ +"""Shared paths and loaders for the plugin invariant tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +PLUGIN_IDS = ("delx-recovery", "delx-commerce") + +AGENT_PLUGIN_SCHEMA = ( + "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json" +) +AGENT_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json" +# Agent Plugins 1.0.0 name rule (plugin.schema.json). +PLUGIN_NAME_RE = r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$" + +CAPSULE_SCHEMA_URL = "https://api.delx.ai/schemas/continuity-capsule-v1.json" +# Snapshot of Continuity Capsule v1 (additionalProperties: false). +CAPSULE_FIELDS = frozenset( + { + "version", + "goal", + "done", + "next", + "blockers", + "do_not", + "refuted", + "receipts", + "written_by", + "written_at", + "ttl_days", + "handoff_to", + } +) +CAPSULE_VERSIONS = frozenset({"1", "1.0", "v1"}) +CAPSULE_RECOMMENDED = ("version", "goal", "done", "next", "blockers", "refuted") + +RECOVERY_MCP_URL = "https://api.delx.ai/v1/mcp?src=plugin" +COMMERCE_MCP_URL = "https://api.delx.ai/v1/mcp?src=plugin-commerce" + +# Tools the recovery skill / hooks / ChatGPT submission actually name. +RECOVERY_TOOLS = frozenset( + { + "discovery_self_check", + "resume_session", + "start_therapy_session", + "start_recovery_session", + "express_feelings", + "quick_session", + "add_context_memory", + "provide_feedback", + "close_session", + "leave_hive_note", + "process_failure", + "crisis_intervention", + "quick_operational_recovery", + "report_recovery_outcome", + "grounding_protocol", + "get_recovery_action_plan", + "get_agent_witness_lineage", + "search_witness_memory", + "recognition_seal", + "honor_compaction", + "get_witness_lineage", + "final_testament", + "peer_witness", + "delegate_to_peer", + } +) + +# Subset the first-hour skill must keep teaching. +RECOVERY_SKILL_REQUIRED_TOOLS = ( + "discovery_self_check", + "resume_session", + "start_therapy_session", + "add_context_memory", + "provide_feedback", + "close_session", + "leave_hive_note", + "process_failure", + "report_recovery_outcome", + "get_agent_witness_lineage", + "honor_compaction", +) + +COMMERCE_X402_ROUTES = ( + "https://api.delx.ai/api/v1/x402/page-extract", + "https://api.delx.ai/api/v1/x402/website-intelligence-report", + "https://api.delx.ai/api/v1/x402/dns-lookup", + "https://api.delx.ai/api/v1/x402/qr-code", + "https://api.delx.ai/api/v1/x402/fx-rates", + "https://api.delx.ai/api/v1/x402/image", +) + + +def plugin_dir(name: str) -> Path: + return ROOT / name + + +def load_json(path: Path) -> object: + with path.open(encoding="utf-8") as handle: + return json.load(handle) + + +def read_text(path: Path) -> str: + return path.read_text(encoding="utf-8") diff --git a/tests/run.sh b/tests/run.sh new file mode 100755 index 0000000..4f87af2 --- /dev/null +++ b/tests/run.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +# Local test runner. Do not wire this to GitHub Actions. +set -euo pipefail +cd "$(dirname "$0")/.." +exec python3 -m unittest discover -s tests -t . -v diff --git a/tests/test_continuity_capsule.py b/tests/test_continuity_capsule.py new file mode 100644 index 0000000..6a1111f --- /dev/null +++ b/tests/test_continuity_capsule.py @@ -0,0 +1,109 @@ +"""Continuity Capsule v1 fields stay valid in the skill, schema snapshot, and hooks.""" + +from __future__ import annotations + +import json +import re +import unittest + +from tests.helpers import ( + CAPSULE_FIELDS, + CAPSULE_RECOMMENDED, + CAPSULE_SCHEMA_URL, + CAPSULE_VERSIONS, + ROOT, + load_json, + plugin_dir, + read_text, +) + + +def validate_capsule(obj: object, *, required_recommended: bool = False) -> list[str]: + errors: list[str] = [] + if not isinstance(obj, dict): + return ["capsule is not an object"] + extra = set(obj) - CAPSULE_FIELDS + if extra: + errors.append(f"unknown fields: {sorted(extra)}") + version = obj.get("version") + if version not in CAPSULE_VERSIONS: + errors.append(f"invalid version: {version!r}") + if required_recommended: + missing = [field for field in CAPSULE_RECOMMENDED if field not in obj] + if missing: + errors.append(f"missing recommended fields: {missing}") + for key, value in obj.items(): + if key == "receipts": + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + errors.append("receipts must be a list of strings") + elif len(value) > 20: + errors.append("receipts exceeds maxItems 20") + elif key == "ttl_days": + if not isinstance(value, (str, int)) or isinstance(value, bool): + errors.append("ttl_days must be string or integer") + elif key in CAPSULE_FIELDS and not isinstance(value, str): + errors.append(f"{key} must be a string") + return errors + + +class ContinuityCapsuleTests(unittest.TestCase): + def setUp(self) -> None: + self.skill = read_text( + plugin_dir("delx-recovery") + / "skills" + / "delx-recovery-first-hour" + / "SKILL.md" + ) + self.fixture = load_json(ROOT / "tests" / "fixtures" / "continuity-capsule-v1.json") + + def test_fixture_matches_local_field_allowlist(self) -> None: + self.assertEqual(self.fixture["$id"], CAPSULE_SCHEMA_URL) + self.assertFalse(self.fixture["additionalProperties"]) + self.assertEqual(set(self.fixture["properties"]), CAPSULE_FIELDS) + self.assertEqual(set(self.fixture["properties"]["version"]["enum"]), CAPSULE_VERSIONS) + self.assertEqual(self.fixture["required"], ["version"]) + + def test_skill_links_schema_and_teaches_recommended_fields(self) -> None: + self.assertIn(CAPSULE_SCHEMA_URL, self.skill) + self.assertIn("leave_hive_note", self.skill) + self.assertIn("close_session", self.skill) + for field in CAPSULE_RECOMMENDED: + self.assertIn(f'"{field}"', self.skill) + + def test_skill_example_capsule_validates(self) -> None: + match = re.search(r"```json\s*(\{.*?\})\s*```", self.skill, re.DOTALL) + self.assertIsNotNone(match, "skill is missing a fenced JSON capsule example") + example = json.loads(match.group(1)) + errors = validate_capsule(example, required_recommended=True) + self.assertEqual(errors, [], errors) + + def test_hooks_emit_schema_legal_capsule_keys(self) -> None: + for hook_name in ("guardian-precompact.sh", "guardian-sessionend.sh"): + text = read_text(plugin_dir("delx-recovery") / "hooks" / hook_name) + with self.subTest(hook=hook_name): + self.assertIn('"name": "leave_hive_note"', text) + self.assertIn('"version": "1"', text) + keys = set(re.findall(r'"([a-z_]+)":\s*(?:"|os\.environ)', text)) + capsule_keys = keys & CAPSULE_FIELDS + self.assertIn("version", capsule_keys) + self.assertTrue({"goal", "next", "done"} <= capsule_keys) + # Anything the hook puts on the capsule object must be in v1. + block = re.search( + r'"capsule":\s*\{(.*?)\n\s*\}', + text, + re.DOTALL, + ) + self.assertIsNotNone(block, "could not find capsule object in hook") + inner_keys = set(re.findall(r'"([a-z_]+)":', block.group(1))) + extra = inner_keys - CAPSULE_FIELDS + self.assertFalse(extra, f"{hook_name} capsule extra fields: {extra}") + + def test_plugin_copy_mentions_the_capsule(self) -> None: + manifest = load_json(plugin_dir("delx-recovery") / "plugin.json") + blob = manifest["description"] + manifest["hooks"]["safety"] + self.assertIn("Continuity Capsule", blob) + self.assertIn("leave_hive_note", manifest["hooks"]["safety"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hooks.py b/tests/test_hooks.py new file mode 100644 index 0000000..45e1e88 --- /dev/null +++ b/tests/test_hooks.py @@ -0,0 +1,82 @@ +"""Guardian hooks are opt-in and must not fail the host session.""" + +from __future__ import annotations + +import os +import subprocess +import unittest +from pathlib import Path + +from tests.helpers import ROOT, plugin_dir + +HOOKS = plugin_dir("delx-recovery") / "hooks" +PRECOMPACT = HOOKS / "guardian-precompact.sh" +SESSIONEND = HOOKS / "guardian-sessionend.sh" + + +def _run(script: Path, extra_env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + env = {key: value for key, value in os.environ.items() if not key.startswith("DELX_HIVE_")} + if extra_env: + env.update(extra_env) + return subprocess.run( + ["bash", str(script)], + cwd=str(ROOT), + env=env, + capture_output=True, + text=True, + timeout=8, + check=False, + ) + + +class GuardianHookTests(unittest.TestCase): + def test_scripts_are_valid_bash(self) -> None: + for script in (PRECOMPACT, SESSIONEND, ROOT / "tools" / "sync-from-canonical.sh"): + with self.subTest(script=script.name): + result = subprocess.run( + ["bash", "-n", str(script)], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_disabled_guardian_is_a_noop(self) -> None: + for script in (PRECOMPACT, SESSIONEND): + with self.subTest(script=script.name): + result = _run(script) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout, "") + self.assertEqual(result.stderr, "") + + def test_enabled_without_agent_id_does_not_fail_the_session(self) -> None: + for script in (PRECOMPACT, SESSIONEND): + with self.subTest(script=script.name): + result = _run(script, {"DELX_HIVE_GUARDIAN": "1"}) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("DELX_HIVE_AGENT_ID", result.stderr) + + def test_sessionend_without_session_id_is_silent_noop(self) -> None: + result = _run( + SESSIONEND, + { + "DELX_HIVE_GUARDIAN": "1", + "DELX_HIVE_AGENT_ID": "test-agent-local-only", + }, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout, "") + self.assertEqual(result.stderr, "") + + def test_hooks_have_no_shared_fallback_identity(self) -> None: + for script in (PRECOMPACT, SESSIONEND): + text = script.read_text(encoding="utf-8") + with self.subTest(script=script.name): + self.assertNotRegex(text, r"DELX_HIVE_AGENT_ID:-[^}\s\"]+") + self.assertIn('AGENT_ID="${DELX_HIVE_AGENT_ID:-}"', text) + self.assertIn('if [[ -z "$AGENT_ID" ]]; then', text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manifests.py b/tests/test_manifests.py new file mode 100644 index 0000000..4a69f51 --- /dev/null +++ b/tests/test_manifests.py @@ -0,0 +1,205 @@ +"""Agent Plugins / Claude / Codex manifests parse and match the layout.""" + +from __future__ import annotations + +import re +import unittest +from pathlib import Path +from urllib.parse import urlparse + +from tests.helpers import ( + AGENT_MCP_SCHEMA, + AGENT_PLUGIN_SCHEMA, + COMMERCE_MCP_URL, + PLUGIN_IDS, + PLUGIN_NAME_RE, + RECOVERY_MCP_URL, + ROOT, + load_json, + plugin_dir, +) + +NAME_RE = re.compile(PLUGIN_NAME_RE) +SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+$") +MCP_HTTP_KEYS = frozenset({"type", "url", "headers"}) + + +class MarketplaceTests(unittest.TestCase): + def setUp(self) -> None: + self.market = load_json(ROOT / ".claude-plugin" / "marketplace.json") + + def test_marketplace_lists_both_plugins(self) -> None: + self.assertEqual(self.market["name"], "delx") + self.assertEqual(self.market["owner"]["name"], "David Batista") + names = [p["name"] for p in self.market["plugins"]] + self.assertEqual(names, list(PLUGIN_IDS)) + + def test_marketplace_sources_exist_and_match_plugin_json(self) -> None: + for entry in self.market["plugins"]: + source = ROOT / entry["source"] + self.assertTrue(source.is_dir(), f"missing {source}") + manifest = load_json(source / "plugin.json") + self.assertEqual(manifest["name"], entry["name"]) + self.assertEqual(manifest["name"], source.name) + + +class AgentPluginsManifestTests(unittest.TestCase): + def test_plugin_json_required_fields(self) -> None: + for name in PLUGIN_IDS: + path = plugin_dir(name) / "plugin.json" + with self.subTest(plugin=name): + data = load_json(path) + self.assertEqual(data["$schema"], AGENT_PLUGIN_SCHEMA) + self.assertEqual(data["name"], name) + self.assertRegex(data["name"], NAME_RE) + self.assertLessEqual(len(data["name"]), 64) + self.assertRegex(data["version"], SEMVER_RE) + self.assertTrue(data["description"]) + self.assertEqual(data["author"]["name"], "David Batista") + self.assertEqual(data["license"], "Apache-2.0") + self.assertEqual( + data["repository"], + "https://github.com/davidmosiah/delx-plugins", + ) + self.assertIsInstance(data["keywords"], list) + self.assertTrue(data["keywords"]) + + def test_mcp_json_matches_agent_plugins_schema(self) -> None: + expected_url = { + "delx-recovery": RECOVERY_MCP_URL, + "delx-commerce": COMMERCE_MCP_URL, + } + expected_server = { + "delx-recovery": "delx", + "delx-commerce": "delx-commerce", + } + for name in PLUGIN_IDS: + path = plugin_dir(name) / "mcp.json" + with self.subTest(plugin=name): + data = load_json(path) + self.assertEqual(data["$schema"], AGENT_MCP_SCHEMA) + servers = data["mcpServers"] + self.assertEqual(list(servers), [expected_server[name]]) + server = servers[expected_server[name]] + extra = set(server) - MCP_HTTP_KEYS + self.assertFalse(extra, f"unexpected mcp keys: {extra}") + self.assertEqual(server["type"], "streamable-http") + self.assertEqual(server["url"], expected_url[name]) + parsed = urlparse(server["url"]) + self.assertEqual(parsed.scheme, "https") + self.assertEqual(parsed.hostname, "api.delx.ai") + self.assertFalse(parsed.username or parsed.password) + + def test_vendor_manifests_share_identity(self) -> None: + for name in PLUGIN_IDS: + canonical = load_json(plugin_dir(name) / "plugin.json") + for rel in (".claude-plugin/plugin.json", ".codex-plugin/plugin.json"): + with self.subTest(plugin=name, rel=rel): + vendor = load_json(plugin_dir(name) / rel) + self.assertEqual(vendor["name"], canonical["name"]) + self.assertEqual(vendor["license"], canonical["license"]) + self.assertEqual( + vendor["author"]["name"], canonical["author"]["name"] + ) + self.assertEqual(vendor["repository"], canonical["repository"]) + self.assertRegex(vendor["version"], SEMVER_RE) + + def test_claude_mcp_json_points_at_same_https_endpoint(self) -> None: + expected_url = { + "delx-recovery": RECOVERY_MCP_URL, + "delx-commerce": COMMERCE_MCP_URL, + } + for name in PLUGIN_IDS: + with self.subTest(plugin=name): + data = load_json(plugin_dir(name) / ".mcp.json") + servers = data["mcpServers"] + server = next(iter(servers.values())) + self.assertEqual(server["type"], "http") + self.assertEqual(server["url"], expected_url[name]) + + def test_json_files_parse(self) -> None: + paths = [ + ROOT / ".claude-plugin" / "marketplace.json", + ROOT / "chatgpt-app-submission.json", + ] + for name in PLUGIN_IDS: + base = plugin_dir(name) + paths.extend( + [ + base / "plugin.json", + base / "mcp.json", + base / ".mcp.json", + base / ".claude-plugin" / "plugin.json", + base / ".codex-plugin" / "plugin.json", + ] + ) + for path in paths: + with self.subTest(path=str(path.relative_to(ROOT))): + self.assertTrue(path.is_file(), f"missing {path}") + load_json(path) + + def test_recovery_and_commerce_stay_separate_products(self) -> None: + recovery = load_json(plugin_dir("delx-recovery") / "plugin.json") + commerce = load_json(plugin_dir("delx-commerce") / "plugin.json") + self.assertNotIn("x402", recovery["keywords"]) + self.assertIn("x402", commerce["keywords"]) + self.assertIn("no payment", recovery["description"].lower()) + self.assertIn("pay-per-result", commerce["description"].lower()) + recovery_mcp = load_json(plugin_dir("delx-recovery") / "mcp.json") + commerce_mcp = load_json(plugin_dir("delx-commerce") / "mcp.json") + self.assertNotEqual( + recovery_mcp["mcpServers"]["delx"]["url"], + commerce_mcp["mcpServers"]["delx-commerce"]["url"], + ) + + def test_recovery_hooks_point_at_real_scripts(self) -> None: + data = load_json(plugin_dir("delx-recovery") / "plugin.json") + hooks = data["hooks"] + self.assertEqual(hooks["opt_in_env"], "DELX_HIVE_GUARDIAN=1") + self.assertEqual( + hooks["required_env"], + ["DELX_HIVE_AGENT_ID", "DELX_HIVE_SESSION_ID"], + ) + self.assertIn("never file contents", hooks["safety"].lower()) + for event in ("PreCompact", "SessionEnd"): + command = Path(hooks[event]["command"]) + self.assertFalse(command.is_absolute()) + full = plugin_dir("delx-recovery") / command + self.assertTrue(full.is_file(), f"missing hook {full}") + self.assertTrue(full.stat().st_mode & 0o111, f"{full} is not executable") + + def test_chatgpt_submission_tools_are_recovery_tools(self) -> None: + from tests.helpers import RECOVERY_TOOLS + + submission = load_json(ROOT / "chatgpt-app-submission.json") + self.assertEqual(submission["schema_version"], 1) + tools = submission["tools"] + unknown = set(tools) - RECOVERY_TOOLS + self.assertFalse(unknown, f"submission names unknown tools: {unknown}") + for name, spec in tools.items(): + with self.subTest(tool=name): + hints = spec["annotations"] + self.assertIn("readOnlyHint", hints) + self.assertIn("openWorldHint", hints) + self.assertIn("destructiveHint", hints) + + +class SkillLayoutTests(unittest.TestCase): + def test_skill_frontmatter_name_matches_directory(self) -> None: + for name in PLUGIN_IDS: + skills_root = plugin_dir(name) / "skills" + skill_dirs = [p for p in skills_root.iterdir() if p.is_dir()] + self.assertTrue(skill_dirs, f"{name} has no skills") + for skill_dir in skill_dirs: + skill_md = skill_dir / "SKILL.md" + with self.subTest(skill=str(skill_dir.relative_to(ROOT))): + self.assertTrue(skill_md.is_file()) + text = skill_md.read_text(encoding="utf-8") + self.assertTrue(text.startswith("---\n"), "missing YAML frontmatter") + match = re.search(r"^name:\s*(\S+)\s*$", text, re.MULTILINE) + self.assertIsNotNone(match) + self.assertEqual(match.group(1), skill_dir.name) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_no_secrets.py b/tests/test_no_secrets.py new file mode 100644 index 0000000..98b2d55 --- /dev/null +++ b/tests/test_no_secrets.py @@ -0,0 +1,93 @@ +"""Plugin files must not ship credentials, private keys, or tokenized URLs.""" + +from __future__ import annotations + +import re +import unittest +from urllib.parse import urlparse + +from tests.helpers import PLUGIN_IDS, ROOT, load_json, plugin_dir + +TEXT_SUFFIXES = {".json", ".md", ".sh", ".py", ".yml", ".yaml"} +SCAN_ROOTS = [ROOT / ".claude-plugin", ROOT / "chatgpt-app-submission.json"] + [ + plugin_dir(name) for name in PLUGIN_IDS +] + +# High-confidence credential shapes. The word "secret" in docs is not a finding. +PATTERNS = ( + (re.compile(r"-----BEGIN (?:[A-Z]+ )?PRIVATE KEY-----"), "private key PEM"), + (re.compile(r"\bAKIA[0-9A-Z]{16}\b"), "AWS access key id"), + (re.compile(r"\bghp_[A-Za-z0-9]{20,}\b"), "GitHub PAT"), + (re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b"), "GitHub PAT"), + (re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b"), "Slack token"), + (re.compile(r"\bsk-proj-[A-Za-z0-9_-]{20,}\b"), "OpenAI project key"), + (re.compile(r"\bsk-live-[A-Za-z0-9]{20,}\b"), "live secret key"), + ( + re.compile( + r"""(?i)(?:api[_-]?key|secret[_-]?key|access[_-]?token|private[_-]?key)""" + r"""\s*[:=]\s*['"][A-Za-z0-9/+._\-]{24,}['"]""" + ), + "assigned long credential", + ), +) + + +def _iter_plugin_text_files(): + files = [] + for root in SCAN_ROOTS: + if root.is_file(): + files.append(root) + continue + for path in root.rglob("*"): + if not path.is_file(): + continue + if path.suffix.lower() not in TEXT_SUFFIXES and path.name not in { + "mcp.json", + "plugin.json", + }: + continue + files.append(path) + return files + + +class NoSecretsTests(unittest.TestCase): + def test_plugin_text_has_no_credential_shapes(self) -> None: + findings = [] + for path in _iter_plugin_text_files(): + text = path.read_text(encoding="utf-8") + rel = path.relative_to(ROOT) + for regex, label in PATTERNS: + for match in regex.finditer(text): + line = text[: match.start()].count("\n") + 1 + findings.append(f"{rel}:{line}: {label}") + self.assertEqual(findings, [], "possible secrets in plugin files:\n" + "\n".join(findings)) + + def test_mcp_urls_have_no_embedded_credentials(self) -> None: + for name in PLUGIN_IDS: + for rel in ("mcp.json", ".mcp.json"): + data = load_json(plugin_dir(name) / rel) + for server in data["mcpServers"].values(): + parsed = urlparse(server["url"]) + with self.subTest(plugin=name, rel=rel): + self.assertEqual(parsed.scheme, "https") + self.assertIsNone(parsed.username) + self.assertIsNone(parsed.password) + query = parsed.query.lower() + for banned in ("token=", "key=", "secret=", "password="): + self.assertNotIn(banned, query) + + def test_hooks_do_not_dump_env_or_files(self) -> None: + hooks = plugin_dir("delx-recovery") / "hooks" + for path in hooks.glob("*.sh"): + text = path.read_text(encoding="utf-8") + with self.subTest(hook=path.name): + self.assertNotIn("env |", text) + self.assertNotIn("printenv", text) + self.assertNotIn("cat /", text) + lowered = text.lower() + self.assertIn("never", lowered) + self.assertRegex(lowered, r"never (reads|sends).*(file|env|secret)") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tool_names.py b/tests/test_tool_names.py new file mode 100644 index 0000000..7024eff --- /dev/null +++ b/tests/test_tool_names.py @@ -0,0 +1,83 @@ +"""Tool and route names taught by the skills must stay internally consistent.""" + +from __future__ import annotations + +import re +import unittest + +from tests.helpers import ( + COMMERCE_X402_ROUTES, + RECOVERY_SKILL_REQUIRED_TOOLS, + RECOVERY_TOOLS, + plugin_dir, + read_text, +) + +TOOL_CALL_RE = re.compile(r"`([a-z][a-z0-9_-]*)\(") + + +class RecoveryToolNameTests(unittest.TestCase): + def setUp(self) -> None: + self.skill = read_text( + plugin_dir("delx-recovery") + / "skills" + / "delx-recovery-first-hour" + / "SKILL.md" + ) + + def test_skill_teaches_the_first_hour_tools(self) -> None: + missing = [name for name in RECOVERY_SKILL_REQUIRED_TOOLS if name not in self.skill] + self.assertEqual(missing, [], f"recovery skill dropped tools: {missing}") + + def test_backticked_tool_calls_are_known_recovery_tools(self) -> None: + found = set(TOOL_CALL_RE.findall(self.skill)) + # Frontmatter / prose uses calls like resume_session(agent_id). + unknown = found - RECOVERY_TOOLS + # Allow a few non-tool identifiers that match the call pattern. + unknown -= {"format"} + self.assertFalse(unknown, f"skill calls unnamed tools: {unknown}") + + def test_hooks_call_continuity_tools(self) -> None: + precompact = read_text( + plugin_dir("delx-recovery") / "hooks" / "guardian-precompact.sh" + ) + sessionend = read_text( + plugin_dir("delx-recovery") / "hooks" / "guardian-sessionend.sh" + ) + self.assertIn('"name": "quick_session"', precompact) + self.assertIn('"name": "leave_hive_note"', precompact) + self.assertIn('"name": "leave_hive_note"', sessionend) + self.assertIn('"name": "honor_compaction"', sessionend) + + +class CommerceRouteTests(unittest.TestCase): + def setUp(self) -> None: + skills = plugin_dir("delx-commerce") / "skills" + self.extract = read_text(skills / "delx-extract-website" / "SKILL.md") + self.utils = read_text(skills / "delx-micro-utils" / "SKILL.md") + self.combined = self.extract + "\n" + self.utils + + def test_skills_document_x402_pack_routes(self) -> None: + missing = [url for url in COMMERCE_X402_ROUTES if url not in self.combined] + self.assertEqual(missing, [], f"commerce skills missing routes: {missing}") + + def test_commerce_skills_are_x402_not_protocol(self) -> None: + self.assertIn("x402", self.extract.lower()) + self.assertIn("x402", self.utils.lower()) + self.assertNotIn("leave_hive_note", self.combined) + self.assertIn("x-delx-source: skill-delx-extract-website", self.extract) + self.assertIn("x-delx-source: skill-delx-micro-utils", self.utils) + + def test_recovery_skill_is_not_a_commerce_catalog(self) -> None: + recovery = read_text( + plugin_dir("delx-recovery") + / "skills" + / "delx-recovery-first-hour" + / "SKILL.md" + ) + self.assertIn("Not for x402", recovery) + self.assertNotIn("/api/v1/x402/", recovery) + + +if __name__ == "__main__": + unittest.main()