#!/usr/bin/env python3 """lib/mcp_registry.py: one registry, reconciled into every account, for both providers.""" import contextlib import importlib.util import io import json import os from pathlib import Path import stat import subprocess import sys import tempfile import unittest REPO = Path(__file__).resolve().parents[1] LIB = REPO / "lib" / "mcp_registry.py" HAS_TOMLLIB = sys.version_info >= (3, 11) def _load(): spec = importlib.util.spec_from_file_location("mcp_registry", LIB) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module R = _load() STDIO = {"type": "stdio", "command": "npx", "args": ["-y", "appinspire-mcp@latest", "serve"], "env": {}} HTTP = {"type": "http", "url": "https://mcp.example/mcp", "headers": {"Authorization": "Bearer ${TOKEN}", "X-Plain": "1", "X-Env": "${OTHER}"}} class RegistryCase(unittest.TestCase): def setUp(self): self.work = tempfile.TemporaryDirectory() self.addCleanup(self.work.cleanup) self.root = Path(self.work.name) / "pool" self.root.mkdir() (self.root / "accounts.json").write_text(json.dumps({ "version": 1, "accounts": [{"id": "acct-01"}, {"id": "acct-02"}, {"id": "acct-99"}]})) for account in ("acct-01", "acct-02"): (self.root / account).mkdir() def acct(self, account="acct-01"): return str(self.root / account) def run_cli(self, provider, *argv, stdin=None): out, err = io.StringIO(), io.StringIO() old_stdin = sys.stdin if stdin is not None: sys.stdin = io.StringIO(stdin) try: with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): rc = R.main(["--root", str(self.root), "--provider", provider, *argv]) finally: sys.stdin = old_stdin return rc, out.getvalue(), err.getvalue() def claude_doc(self, account="acct-01"): return json.loads((self.root / account / ".claude.json").read_text()) def registry(self): return json.loads((self.root / "mcp-servers.json").read_text()) def overlay(self): return json.loads((self.root / "mcp-servers.local.json").read_text()) class EffectiveMergeTests(RegistryCase): def test_registry_wins_over_overlay_both_ways(self): registry = R._normalize_section({"mcpServers": {"a": STDIO}, "retired": ["b"]}) overlay = {"owners": {"robot": R._normalize_section({"mcpServers": {"b": STDIO, "c": STDIO}, "retired": ["a", "d"]})}} eff = R.effective(registry, overlay) self.assertEqual(sorted(eff.servers), ["a", "c"]) # a survives the overlay tombstone self.assertEqual(eff.retired, {"b", "d"}) # b is gone despite the overlay self.assertNotIn("a", eff.retired) def test_overlay_tombstone_beats_overlay_server_and_user_scope_is_kept_apart(self): overlay = {"owners": {"one": R._normalize_section({"mcpServers": {"x": STDIO, "y": STDIO}}), "two": R._normalize_section({"retired": ["x"], "retiredUser": ["y"]})}} eff = R.effective(R.load_registry(str(self.root)), overlay) self.assertEqual(eff.servers, {}) self.assertEqual(eff.retired, {"x"}) self.assertEqual(eff.retired_user, {"y"}) class ClaudeReconcileTests(RegistryCase): def test_upsert_purges_tombstones_everywhere_and_keeps_extras(self): (self.root / "acct-01" / ".claude.json").write_text(json.dumps({ "numStartups": 3, "mcpServers": {"adspower-local-api": STDIO, "mine": STDIO, "appinspire-mcp": {"type": "stdio", "command": "old", "args": [], "env": {}}}, "projects": {"/p": {"mcpServers": {"adspower-local-api": STDIO, "keep": STDIO}}}, })) rc, _, _ = self.run_cli("claude", "add", "appinspire-mcp", "-e", "K=V", "--", "npx", "-y", "appinspire-mcp@latest", "serve") self.assertEqual(rc, 0) rc, _, _ = self.run_cli("claude", "remove", "adspower-local-api") self.assertEqual(rc, 0) doc = self.claude_doc() self.assertEqual(doc["numStartups"], 3) self.assertEqual(sorted(doc["mcpServers"]), ["appinspire-mcp", "mine"]) self.assertEqual(doc["mcpServers"]["appinspire-mcp"], {"type": "stdio", "command": "npx", "args": ["-y", "appinspire-mcp@latest", "serve"], "env": {"K": "V"}}) self.assertEqual(sorted(doc["projects"]["/p"]["mcpServers"]), ["keep"]) self.assertEqual(sorted(self.claude_doc("acct-02")["mcpServers"]), ["appinspire-mcp"]) self.assertEqual(self.registry()["retired"], ["adspower-local-api"]) self.assertTrue((self.root / "acct-01" / ".mcp-applied").exists()) def test_user_only_tombstone_leaves_project_entries_alone(self): (self.root / "acct-01" / ".claude.json").write_text(json.dumps({ "mcpServers": {"pg": STDIO}, "projects": {"/w": {"mcpServers": {"pg": STDIO}}}})) rc, _, _ = self.run_cli("claude", "remove", "pg", "--user-only") self.assertEqual(rc, 0) doc = self.claude_doc() self.assertNotIn("pg", doc["mcpServers"]) self.assertIn("pg", doc["projects"]["/w"]["mcpServers"]) self.assertEqual(self.registry()["retiredUser"], ["pg"]) self.assertEqual(self.registry()["retired"], []) # a full remove afterwards moves it to the everywhere list and purges the project too self.run_cli("claude", "remove", "pg") self.assertEqual((self.registry()["retired"], self.registry()["retiredUser"]), (["pg"], [])) self.assertNotIn("pg", self.claude_doc()["projects"]["/w"]["mcpServers"]) def test_reseed_is_idempotent_and_re_adding_lifts_the_tombstone(self): self.run_cli("claude", "remove", "gone") rc, out, _ = self.run_cli("claude", "apply", "--all") self.assertEqual(rc, 0) before = (self.root / "acct-01" / ".claude.json").read_text() rc, out, _ = self.run_cli("claude", "apply", "--all") self.assertIn("acct-01: unchanged", out) self.assertEqual(before, (self.root / "acct-01" / ".claude.json").read_text()) self.run_cli("claude", "add", "gone", "--", "echo") self.assertEqual(self.registry()["retired"], []) self.assertIn("gone", self.claude_doc()["mcpServers"]) def test_project_scope_creates_the_client_skeleton_and_unions_approvals(self): rc, _, _ = self.run_cli("claude", "add", "appinspire", "--scope", "project", "--project", "/Users/x/lab", "-e", "APPINSPIRE_LIBRARY_DIR=/lib", "--", "node", "/src/bin/appinspire-mcp.mjs", "serve") self.assertEqual(rc, 0) entry = self.claude_doc()["projects"]["/Users/x/lab"] for key in R.PROJECT_SKELETON: self.assertIn(key, entry) self.assertEqual(entry["mcpServers"]["appinspire"]["env"], {"APPINSPIRE_LIBRARY_DIR": "/lib"}) self.assertNotIn("appinspire", self.claude_doc()["mcpServers"]) registry = self.registry() registry["projects"]["/Users/x/lab"]["enabledMcpjsonServers"] = ["pj"] (self.root / "mcp-servers.json").write_text(json.dumps(registry)) (self.root / "acct-02" / ".claude.json").write_text(json.dumps({"projects": { "/Users/x/lab": {"mcpServers": {}, "enabledMcpjsonServers": ["own"], "disabledMcpjsonServers": ["pj"]}}})) self.run_cli("claude", "apply", "--all") entry = self.claude_doc("acct-02")["projects"]["/Users/x/lab"] self.assertEqual(entry["enabledMcpjsonServers"], ["own", "pj"]) self.assertEqual(entry["disabledMcpjsonServers"], []) rc, _, _ = self.run_cli("claude", "remove", "appinspire", "--scope", "project", "--project", "/Users/x/lab") self.assertNotIn("appinspire", self.claude_doc()["projects"]["/Users/x/lab"]["mcpServers"]) def test_project_paths_are_normalized_like_the_client_keys_them(self): home = Path(self.work.name) / "home" (home / "lab").mkdir(parents=True) os.environ["HOME"] = str(home) try: rc, _, _ = self.run_cli("claude", "add", "x", "--scope", "project", "--project", "~/lab", "--", "echo") finally: os.environ.pop("HOME", None) self.assertEqual(rc, 0) self.assertEqual(list(self.registry()["projects"]), [os.path.realpath(str(home / "lab"))]) def test_codex_only_keys_never_reach_claude_and_types_are_kept_or_inferred(self): block = {"command": "npx", "args": ["x"], "startup_timeout_sec": 90, "env_vars": ["A"], "codex": {"oauth": {"client_id": "c"}}} rc, _, _ = self.run_cli("claude", "add-json", "svc", json.dumps(block)) self.assertEqual(rc, 0) self.assertEqual(self.claude_doc()["mcpServers"]["svc"], {"type": "stdio", "command": "npx", "args": ["x"], "env": {}}) self.run_cli("claude", "add-json", "web", json.dumps({"url": "https://h/mcp"})) self.assertEqual(self.claude_doc()["mcpServers"]["web"], {"type": "http", "url": "https://h/mcp"}) self.run_cli("claude", "add-json", "sock", json.dumps({"type": "ws", "url": "wss://h/ws"})) self.assertEqual(self.claude_doc()["mcpServers"]["sock"], {"type": "ws", "url": "wss://h/ws"}) def test_http_server_from_the_cli_in_every_argument_order(self): rc, _, _ = self.run_cli("claude", "add", "inspo", "--transport", "http", "https://inspomcp.dev/api/mcp", "-H", "X-Key: abc") self.assertEqual(rc, 0) self.assertEqual(self.claude_doc()["mcpServers"]["inspo"], {"type": "http", "url": "https://inspomcp.dev/api/mcp", "headers": {"X-Key": "abc"}}) rc, _, _ = self.run_cli("claude", "add", "two", "--url", "https://two/mcp") self.assertEqual(self.claude_doc()["mcpServers"]["two"], {"type": "http", "url": "https://two/mcp"}) rc, _, _ = self.run_cli("claude", "add", "three", "https://three/mcp", "--transport", "http") self.assertEqual(self.claude_doc()["mcpServers"]["three"]["url"], "https://three/mcp") with self.assertRaises(R.RegistryError): self.run_cli("claude", "add", "four", "https://a", "https://b") def test_writes_are_atomic_and_keep_the_file_mode(self): target = self.root / "acct-01" / ".claude.json" target.write_text("{}") os.chmod(target, 0o640) self.run_cli("claude", "add", "s", "--", "echo") self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o640) self.assertEqual(stat.S_IMODE((self.root / "acct-02" / ".claude.json").stat().st_mode), 0o600) self.assertEqual([p.name for p in (self.root / "acct-01").glob(".mcp-registry-*")], []) def test_a_corrupt_account_document_is_reported_not_overwritten_and_exits_3(self): (self.root / "acct-01" / ".claude.json").write_text("{not json") rc, out, err = self.run_cli("claude", "add", "s", "--", "echo") self.assertEqual(rc, R.EXIT_PARTIAL) self.assertIn("acct-01", err) self.assertIn("mcp apply", err) self.assertEqual((self.root / "acct-01" / ".claude.json").read_text(), "{not json") self.assertIn("acct-02: updated", out) self.assertIn("s", self.registry()["mcpServers"]) # the registry WAS saved def test_missing_manifest_account_dirs_are_skipped(self): rc, out, _ = self.run_cli("claude", "add", "s", "--", "echo") self.assertEqual(rc, 0) self.assertNotIn("acct-99", out) self.assertFalse((self.root / "acct-99").exists()) class ValidationTests(RegistryCase): def test_malformed_blocks_are_refused_before_the_registry_is_saved(self): for block in ({"command": "x", "args": 5}, {"command": "npx", "args": "-y pkg"}, {"command": "x", "enabled": "false"}, {"command": "x", "startup_timeout_sec": "90"}, {"command": "x", "startup_timeout_sec": -1}, {"command": "x", "startup_timeout_sec": 1e300}, {"command": "x", "env": {"A": ["list"]}}, {"type": "carrier", "command": "x"}, {"command": ""}, {"url": 7}): with self.subTest(block=block): with self.assertRaises(R.RegistryError): self.run_cli("claude", "add-json", "bad", json.dumps(block)) self.assertFalse((self.root / "mcp-servers.json").exists()) def test_a_bad_block_that_slipped_in_skips_only_itself(self): self.run_cli("claude", "add", "good", "--", "echo", "ok") registry = self.registry() registry["mcpServers"]["bad"] = {"command": "x", "args": 5} (self.root / "mcp-servers.json").write_text(json.dumps(registry)) rc, out, err = self.run_cli("claude", "apply", "--all") self.assertEqual(rc, 0) self.assertIn("bad", err) self.assertIn("good", self.claude_doc("acct-02")["mcpServers"]) self.assertNotIn("bad", self.claude_doc("acct-02")["mcpServers"]) rc, out, _ = self.run_cli("claude", "list") self.assertEqual(rc, 0) # a listing never crashes on it self.assertIn("bad", out) def test_names_follow_the_clients_rule_and_remove_is_lenient(self): rc, _, _ = self.run_cli("claude", "add", "_dev-1", "--", "echo") self.assertEqual(rc, 0) for bad in ("../x", "my.server", "a b", "", "-"[1:]): with self.subTest(bad=bad): with self.assertRaises(R.RegistryError): self.run_cli("claude", "add", bad, "--", "echo") rc, _, _ = self.run_cli("claude", "remove", "my.server") # learned names can always go self.assertEqual(rc, 0) self.assertIn("my.server", self.registry()["retired"]) with self.assertRaises(R.RegistryError): self.run_cli("claude", "remove", "a/b") def test_bad_json_and_missing_command_are_refused(self): with self.assertRaises(R.RegistryError): self.run_cli("claude", "add-json", "x", "{nope") with self.assertRaises(R.RegistryError): self.run_cli("claude", "add-json", "x", json.dumps({"type": "stdio"})) with self.assertRaises(R.RegistryError): self.run_cli("claude", "add", "x") class StampTests(RegistryCase): def test_signature_matches_the_shim_contract(self): self.run_cli("claude", "add", "s", "--", "echo") registry = self.root / "mcp-servers.json" config = self.root / "acct-01" / ".claude.json" stamp = (self.root / "acct-01" / ".mcp-applied").read_text().strip() expected = ",".join([f"{int(registry.stat().st_mtime)}:{registry.stat().st_size}", "-", f"{int(config.stat().st_mtime)}:{config.stat().st_size}"]) self.assertEqual(stamp, expected) self.assertEqual(R.stamp_signature(str(self.root), "claude", self.acct()), expected) stat_flag = "-f" if sys.platform == "darwin" else "-c" fmt = "%m:%z" if sys.platform == "darwin" else "%Y:%s" shell = subprocess.run(["stat", "-L", stat_flag, fmt, str(registry), str(config)], capture_output=True, text=True, check=True).stdout.split() self.assertEqual(shell, [expected.split(",")[0], expected.split(",")[2]]) class LearnTests(RegistryCase): def test_claude_learns_user_project_and_approval_changes(self): (self.root / "acct-01" / ".claude.json").write_text(json.dumps({ "mcpServers": {"old": STDIO}, "projects": {"/proj": {"mcpServers": {"gone": STDIO}, "enabledMcpjsonServers": []}}})) rc, snap, _ = self.run_cli("claude", "snapshot", "--account-dir", self.acct()) before = self.root / "before.json" before.write_text(snap) doc = self.claude_doc() doc["mcpServers"] = {"learned": HTTP} # old removed, learned added doc["projects"]["/proj"] = {"mcpServers": {"local1": STDIO}, "enabledMcpjsonServers": ["pj"]} (self.root / "acct-01" / ".claude.json").write_text(json.dumps(doc)) rc, _, err = self.run_cli("claude", "learn", "--account-dir", self.acct(), "--before", str(before), "--project", "/proj") self.assertEqual(rc, 0) self.assertIn('mirrored "learned"', err) registry = self.registry() self.assertEqual(registry["mcpServers"], {"learned": HTTP}) self.assertEqual(registry["retiredUser"], ["old"]) self.assertEqual(registry["retired"], []) project = registry["projects"]["/proj"] self.assertEqual(project["mcpServers"], {"local1": STDIO}) self.assertEqual(project["retired"], ["gone"]) self.assertEqual(project["enabledMcpjsonServers"], ["pj"]) other = self.claude_doc("acct-02") self.assertEqual(sorted(other["mcpServers"]), ["learned"]) self.assertEqual(sorted(other["projects"]["/proj"]["mcpServers"]), ["local1"]) self.assertEqual(other["projects"]["/proj"]["enabledMcpjsonServers"], ["pj"]) def test_a_learned_user_scope_removal_never_touches_project_entries(self): (self.root / "acct-01" / ".claude.json").write_text(json.dumps({ "mcpServers": {"pg": STDIO}, "projects": {"/a": {"mcpServers": {"pg": STDIO}}, "/b": {"mcpServers": {"pg": STDIO}}}})) (self.root / "acct-02" / ".claude.json").write_text(json.dumps({ "projects": {"/c": {"mcpServers": {"pg": STDIO}}}})) rc, snap, _ = self.run_cli("claude", "snapshot", "--account-dir", self.acct()) before = self.root / "before.json" before.write_text(snap) doc = self.claude_doc() doc["mcpServers"] = {} # what `claude mcp remove -s user pg` does (self.root / "acct-01" / ".claude.json").write_text(json.dumps(doc)) rc, _, _ = self.run_cli("claude", "learn", "--account-dir", self.acct(), "--before", str(before)) self.assertEqual(rc, 0) self.assertEqual(self.registry()["retiredUser"], ["pg"]) for account, path in (("acct-01", "/a"), ("acct-01", "/b"), ("acct-02", "/c")): self.assertIn("pg", self.claude_doc(account)["projects"][path]["mcpServers"]) self.assertNotIn("pg", self.claude_doc("acct-01")["mcpServers"]) def test_nothing_changed_is_quiet(self): rc, snap, _ = self.run_cli("claude", "snapshot", "--account-dir", self.acct()) before = self.root / "before.json" before.write_text(snap) rc, out, err = self.run_cli("claude", "learn", "--account-dir", self.acct(), "--before", str(before)) self.assertEqual((rc, out, err), (0, "", "")) self.assertFalse((self.root / "mcp-servers.json").exists()) def test_on_a_replica_everything_lands_in_the_overlay(self): (self.root / "sync-role").write_text("replica\n") rc, out, _ = self.run_cli("claude", "add", "here", "--", "echo") self.assertEqual(rc, 0) self.assertIn("overlay (replica pool)", out) self.assertFalse((self.root / "mcp-servers.json").exists()) self.assertEqual(sorted(self.overlay()["owners"]["local"]["mcpServers"]), ["here"]) self.assertIn("here", self.claude_doc("acct-02")["mcpServers"]) rc, snap, _ = self.run_cli("claude", "snapshot", "--account-dir", self.acct()) before = self.root / "before.json" before.write_text(snap) doc = self.claude_doc() doc["mcpServers"]["learned"] = STDIO (self.root / "acct-01" / ".claude.json").write_text(json.dumps(doc)) rc, _, err = self.run_cli("claude", "learn", "--account-dir", self.acct(), "--before", str(before)) self.assertEqual(rc, 0) self.assertIn("replica", err) self.assertIn("learned", self.overlay()["owners"]["local"]["mcpServers"]) self.assertIn("learned", self.claude_doc("acct-02")["mcpServers"]) rc, out, _ = self.run_cli("claude", "remove", "here") self.assertEqual(self.overlay()["owners"]["local"]["retired"], ["here"]) self.assertNotIn("here", self.claude_doc("acct-02")["mcpServers"]) rc, out, _ = self.run_cli("claude", "list", "--json") self.assertTrue(json.loads(out)["replica"]) class OverlayTests(RegistryCase): def import_local(self, owner, payload): return self.run_cli("claude", "import-local", "--owner", owner, stdin=json.dumps(payload)) def test_import_local_owner_sections_and_registry_precedence(self): self.run_cli("claude", "add", "keep", "--", "echo") self.run_cli("claude", "remove", "never") (self.root / "acct-01" / ".claude.json").write_text(json.dumps({"mcpServers": {"adspower-local-api": STDIO}})) rc, _, _ = self.import_local("app-robot", { "mcpServers": {"appinspire-mcp": STDIO, "never": STDIO, "keep": {"command": "other"}}, "retired": ["adspower-local-api", "keep"]}) self.assertEqual(rc, 0) doc = self.claude_doc() self.assertEqual(sorted(doc["mcpServers"]), ["appinspire-mcp", "keep"]) self.assertEqual(doc["mcpServers"]["keep"]["command"], "echo") # registry block, not the overlay's self.assertEqual(sorted(self.overlay()["owners"]), ["app-robot"]) # a second owner adds beside the first without disturbing it rc, _, _ = self.import_local("other", {"mcpServers": {"extra": STDIO}}) self.assertEqual(sorted(self.overlay()["owners"]), ["app-robot", "other"]) self.assertEqual(sorted(self.claude_doc()["mcpServers"]), ["appinspire-mcp", "extra", "keep"]) # re-importing the same section is a no-op for the ops log log_lines = (self.root / "ops.log").read_text().count("import-local") self.import_local("other", {"mcpServers": {"extra": STDIO}}) self.assertEqual((self.root / "ops.log").read_text().count("import-local"), log_lines) def test_an_owner_that_drops_a_server_retires_it(self): self.import_local("app-robot", {"mcpServers": {"appinspire-mcp": STDIO, "tmp": STDIO}}) self.assertIn("tmp", self.claude_doc("acct-02")["mcpServers"]) self.import_local("app-robot", {"mcpServers": {"appinspire-mcp": STDIO}}) self.assertNotIn("tmp", self.claude_doc("acct-02")["mcpServers"]) self.assertEqual(self.overlay()["owners"]["app-robot"]["retired"], ["tmp"]) self.import_local("app-robot", {"mcpServers": {}, "retired": ["appinspire-mcp"]}) self.assertEqual(self.claude_doc("acct-02")["mcpServers"], {}) def test_import_local_rejects_bad_payloads_loudly(self): with self.assertRaises(R.RegistryError): self.import_local("app-robot", {"mcpServers": {"x": {"command": "c", "args": 5}}}) with self.assertRaises(R.RegistryError): self.import_local("bad owner", {"mcpServers": {}}) self.assertFalse((self.root / "mcp-servers.local.json").exists()) class CliGuardTests(RegistryCase): def test_apply_without_a_registry_is_a_silent_no_op(self): rc, out, err = self.run_cli("claude", "apply", "--account-dir", self.acct()) self.assertEqual((rc, out, err), (0, "", "")) self.assertFalse((self.root / "acct-01" / ".claude.json").exists()) def test_fail_open_exits_zero_from_a_subprocess(self): proc = subprocess.run([sys.executable, str(LIB), "--root", "/nonexistent", "--provider", "claude", "apply", "--fail-open", "--account-dir", "/x"], capture_output=True, text=True) self.assertEqual(proc.returncode, 0) self.assertIn("mcp-registry:", proc.stderr) proc = subprocess.run([sys.executable, str(LIB), "--root", "/nonexistent", "--provider", "claude", "apply", "--account-dir", "/x"], capture_output=True, text=True) self.assertEqual(proc.returncode, 1) def test_an_unexpected_python_error_never_escapes_as_a_traceback(self): (self.root / "acct-01" / ".claude.json").write_bytes(b"\xff\xfe{}") proc = subprocess.run([sys.executable, str(LIB), "--root", str(self.root), "--provider", "claude", "add", "s", "--", "echo"], capture_output=True, text=True) self.assertEqual(proc.returncode, R.EXIT_PARTIAL) self.assertNotIn("Traceback", proc.stderr) self.assertIn("acct-02: updated", proc.stdout) def test_list_reports_origin_and_tombstones(self): self.run_cli("claude", "add", "svc", "--", "echo", "hi there") self.run_cli("claude", "remove", "adspower-local-api") self.run_cli("claude", "remove", "pg", "--user-only") rc, out, _ = self.run_cli("claude", "list") self.assertIn("svc", out) self.assertIn("echo 'hi there'", out) self.assertIn("[registry]", out) self.assertIn("retired (removed from every account, every scope): adspower-local-api", out) self.assertIn("retired at user scope only: pg", out) rc, out, _ = self.run_cli("claude", "list", "--json") parsed = json.loads(out) self.assertEqual((parsed["retired"], parsed["retiredUser"]), (["adspower-local-api"], ["pg"])) def test_a_stale_lock_is_broken(self): lock = self.root / ".locks" / "mcp" lock.mkdir(parents=True) os.utime(lock, (0, 0)) rc, _, _ = self.run_cli("claude", "add", "svc", "--", "echo") self.assertEqual(rc, 0) self.assertFalse(lock.exists()) def test_leftover_arguments_are_refused(self): with self.assertRaises(R.RegistryError): self.run_cli("claude", "remove", "x", "stray") with self.assertRaises(R.RegistryError): self.run_cli("claude", "add-json", "x", json.dumps(STDIO), "stray") @unittest.skipUnless(HAS_TOMLLIB, "codex reconcile needs tomllib (Python 3.11+)") class CodexReconcileTests(RegistryCase): def config(self, account="acct-01"): return (self.root / account / "config.toml") def parsed(self, account="acct-01"): import tomllib return tomllib.loads(self.config(account).read_text()) def test_tables_are_appended_and_unrelated_text_is_untouched(self): original = ('model = "gpt-5" # keep this comment\n\n' '[mcp_servers.adspower-local-api]\ncommand = "npx"\nargs = ["-y", "x"]\n\n' '[mcp_servers.adspower-local-api.env]\nPORT = "1"\n\n' '# Feature flags (keep me)\n[features.multi_agent_v2]\nenabled = true\n') self.config().write_text(original) rc, _, _ = self.run_cli("codex", "add", "appinspire-mcp", "-e", "LIT=1", "-e", "SECRET=${SECRET}", "--", "npx", "-y", "appinspire-mcp@latest", "serve") self.assertEqual(rc, 0) rc, _, _ = self.run_cli("codex", "remove", "adspower-local-api") self.assertEqual(rc, 0) text = self.config().read_text() self.assertTrue(text.startswith('model = "gpt-5" # keep this comment\n')) self.assertIn("# Feature flags (keep me)\n[features.multi_agent_v2]\nenabled = true\n", text) self.assertNotIn("adspower", text) self.assertEqual(self.parsed()["mcp_servers"], {"appinspire-mcp": { "command": "npx", "args": ["-y", "appinspire-mcp@latest", "serve"], "env": {"LIT": "1"}, "env_vars": ["SECRET"], "startup_timeout_sec": 60}}) self.assertEqual(sorted(self.parsed("acct-02")), ["mcp_servers"]) before = text rc, out, _ = self.run_cli("codex", "apply", "--all") self.assertIn("acct-01: unchanged", out) self.assertEqual(before, self.config().read_text()) def test_comments_above_the_next_table_survive_an_update(self): self.config().write_text('[mcp_servers.old]\ncommand = "a"\n# belongs to old\n\n' '# Feature flags (keep me)\n[features]\nx = true\n') self.run_cli("codex", "add", "old", "--", "new") text = self.config().read_text() self.assertIn("# Feature flags (keep me)\n[features]\nx = true\n", text) self.assertEqual(self.parsed()["mcp_servers"]["old"]["command"], "new") def test_dotted_keys_under_a_bare_table_are_replaced_too(self): self.config().write_text('[mcp_servers]\nold.command = "a"\nold.args = ["b"]\nother.command = "c"\n') rc, _, _ = self.run_cli("codex", "add", "old", "--", "new") self.assertEqual(rc, 0) doc = self.parsed() self.assertEqual(doc["mcp_servers"]["old"]["command"], "new") self.assertEqual(doc["mcp_servers"]["other"], {"command": "c"}) def test_an_unsupported_layout_is_refused_and_the_file_survives(self): original = 'mcp_servers = { old = { command = "a" } }\n' self.config().write_text(original) rc, out, err = self.run_cli("codex", "add", "old", "--", "new") self.assertEqual(rc, R.EXIT_PARTIAL) self.assertIn("refusing to write config.toml", err) self.assertEqual(self.config().read_text(), original) self.assertIn("acct-02: updated", out) def test_http_headers_env_references_and_inexpressible_transports(self): rc, _, _ = self.run_cli("codex", "add-json", "web", json.dumps(HTTP)) self.assertEqual(rc, 0) self.assertEqual(self.parsed()["mcp_servers"]["web"], {"url": "https://mcp.example/mcp", "http_headers": {"X-Plain": "1"}, "env_http_headers": {"X-Env": "OTHER"}, "bearer_token_env_var": "TOKEN"}) for name, block in (("events", {"type": "sse", "url": "https://s/sse"}), ("sock", {"type": "ws", "url": "wss://s/ws"})): rc, out, err = self.run_cli("codex", "add-json", name, json.dumps(block)) self.assertEqual(rc, 0) self.assertIn("cannot express", err) self.assertNotIn(name, self.config().read_text()) def test_env_references_are_forwarded_only_under_their_own_name(self): block = {"command": "npx", "args": ["-y", "gh"], "env": {"GITHUB_TOKEN": "${GH_TOKEN}", "B": "${B:-dflt}", "OK": "${OK}", "PATH": "/opt/bin:${PATH}"}} rc, out, err = self.run_cli("codex", "add-json", "gh", json.dumps(block)) self.assertEqual(rc, 0) self.assertIn("GITHUB_TOKEN=${GH_TOKEN} cannot be expressed", err) self.assertIn("B=${B:-dflt}", err) table = self.parsed()["mcp_servers"]["gh"] self.assertEqual(table["env_vars"], ["OK"]) self.assertEqual(table["env"], {"PATH": "/opt/bin:${PATH}"}) def test_explicit_codex_keys_and_odd_names_round_trip(self): block = {"command": "node", "args": ["s.mjs"], "startup_timeout_sec": 5, "tool_timeout_sec": 30, "cwd": "/srv", "enabled": False} rc, _, _ = self.run_cli("codex", "add-json", "my_server-1", json.dumps(block)) self.assertEqual(rc, 0) self.assertEqual(self.parsed()["mcp_servers"]["my_server-1"], {"command": "node", "args": ["s.mjs"], "startup_timeout_sec": 5, "tool_timeout_sec": 30, "cwd": "/srv", "enabled": False}) def test_codex_learns_a_stock_add_and_remove_and_keeps_oauth_extras(self): self.config().write_text('[mcp_servers.old]\ncommand = "o"\n') rc, snap, _ = self.run_cli("codex", "snapshot", "--account-dir", self.acct()) before = self.root / "before.json" before.write_text(snap) self.config().write_text('[mcp_servers.new]\ncommand = "echo"\nargs = ["hi"]\nenv_vars = ["TOK"]\n' '[mcp_servers.new.env]\nA = "1"\n' '[mcp_servers.web]\nurl = "http://127.0.0.1:9/mcp"\n' 'oauth_resource = "https://res.example"\nenabled_tools = ["a"]\n' '[mcp_servers.web.oauth]\nclient_id = "cid-123"\ncallback_url = "http://cb"\n') rc, _, err = self.run_cli("codex", "learn", "--account-dir", self.acct(), "--before", str(before)) self.assertEqual(rc, 0) self.assertIn('mirrored "new", "web", "old" (removed)', err) registry = self.registry() self.assertEqual(registry["mcpServers"]["new"], {"type": "stdio", "command": "echo", "args": ["hi"], "env": {"A": "1", "TOK": "${TOK}"}}) self.assertEqual(registry["mcpServers"]["web"]["codex"], {"oauth_resource": "https://res.example", "enabled_tools": ["a"], "oauth": {"client_id": "cid-123", "callback_url": "http://cb"}}) self.assertEqual(registry["retiredUser"], ["old"]) # the source account keeps its OAuth settings, and the other account gets them for account in ("acct-01", "acct-02"): web = self.parsed(account)["mcp_servers"]["web"] self.assertEqual(web["oauth"], {"client_id": "cid-123", "callback_url": "http://cb"}) self.assertEqual(web["oauth_resource"], "https://res.example") self.assertEqual(web["enabled_tools"], ["a"]) doc2 = self.parsed("acct-02") self.assertEqual(doc2["mcp_servers"]["new"]["env_vars"], ["TOK"]) self.assertEqual(doc2["mcp_servers"]["new"]["env"], {"A": "1"}) # nothing to rewrite on a second pass: the passthrough round-trips exactly rc, out, _ = self.run_cli("codex", "apply", "--all") self.assertIn("acct-01: unchanged", out) if __name__ == "__main__": unittest.main()