"""Tests for the codex MCP plugin migration helper.""" from __future__ import annotations from pathlib import Path import pytest from hermes_cli.codex_runtime_plugin_migration import ( MIGRATION_MARKER, MIGRATION_END_MARKER, MigrationReport, _build_hermes_tools_mcp_entry, _format_toml_value, _looks_like_test_tempdir, _strip_existing_managed_block, _strip_unmanaged_plugin_tables, _translate_one_server, migrate, render_codex_toml_section, ) # ---- per-server translation ---- class TestTranslateOneServer: def test_stdio_basic(self): cfg, skipped = _translate_one_server("filesystem", { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"], "env": {"FOO": "bar"}, }) assert cfg == { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"], "env": {"FOO": "bar"}, } assert skipped == [] def test_stdio_with_cwd(self): cfg, _ = _translate_one_server("custom", { "command": "/usr/bin/myserver", "cwd": "/var/lib/mcp", }) assert cfg["cwd"] == "/var/lib/mcp" def test_http_basic(self): cfg, skipped = _translate_one_server("api", { "url": "https://x.example/mcp", "headers": {"Authorization": "Bearer abc"}, }) assert cfg == { "url": "https://x.example/mcp", "http_headers": {"Authorization": "Bearer abc"}, } assert skipped == [] def test_sse_falls_under_streamable_http_with_warning(self): cfg, skipped = _translate_one_server("sse_server", { "url": "http://localhost:8000/sse", "transport": "sse", }) assert cfg["url"] == "http://localhost:8000/sse" assert any("sse" in s.lower() for s in skipped) def test_timeouts_translate(self): cfg, _ = _translate_one_server("x", { "command": "y", "timeout": 180, "connect_timeout": 30, }) assert cfg["tool_timeout_sec"] == 180.0 assert cfg["startup_timeout_sec"] == 30.0 def test_non_numeric_timeout_skipped(self): cfg, skipped = _translate_one_server("x", { "command": "y", "timeout": "not-a-number", }) assert "tool_timeout_sec" not in cfg assert any("timeout" in s and "numeric" in s for s in skipped) def test_disabled_server_emits_enabled_false(self): cfg, _ = _translate_one_server("x", { "command": "y", "enabled": False, }) assert cfg["enabled"] is False def test_enabled_true_omitted(self): cfg, _ = _translate_one_server("x", {"command": "y", "enabled": True}) assert "enabled" not in cfg # codex defaults to true def test_command_and_url_prefers_stdio_warns(self): cfg, skipped = _translate_one_server("x", { "command": "y", "url": "http://z", }) assert "command" in cfg assert "url" not in cfg assert any("url" in s for s in skipped) def test_no_transport_returns_none(self): cfg, skipped = _translate_one_server("broken", {"description": "x"}) assert cfg is None assert "no command or url" in skipped[0] def test_sampling_dropped_with_warning(self): cfg, skipped = _translate_one_server("x", { "command": "y", "sampling": {"enabled": True, "model": "gemini-3-flash"}, }) assert "sampling" not in cfg assert any("sampling" in s for s in skipped) def test_unknown_keys_warned(self): cfg, skipped = _translate_one_server("x", { "command": "y", "totally_made_up_key": "value", }) assert "totally_made_up_key" not in cfg assert any("totally_made_up_key" in s for s in skipped) def test_non_dict_input(self): cfg, skipped = _translate_one_server("x", "notadict") # type: ignore[arg-type] assert cfg is None # ---- TOML rendering ---- class TestTomlValueFormatter: def test_string_quoted(self): assert _format_toml_value("hello") == '"hello"' def test_string_with_quotes_escaped(self): assert _format_toml_value('a"b') == '"a\\"b"' def test_bool(self): assert _format_toml_value(True) == "true" assert _format_toml_value(False) == "false" def test_int(self): assert _format_toml_value(42) == "42" def test_float(self): assert _format_toml_value(180.0) == "180.0" def test_list_of_strings(self): assert _format_toml_value(["a", "b"]) == '["a", "b"]' def test_inline_table(self): out = _format_toml_value({"FOO": "bar"}) assert out == '{ FOO = "bar" }' def test_empty_inline_table(self): assert _format_toml_value({}) == "{}" def test_string_with_newline_escaped(self): """TOML basic strings don't allow literal newlines — a path or env var containing a newline must use \\n. Otherwise codex would refuse to load the config.""" out = _format_toml_value("line one\nline two") assert "\n" not in out # no raw newline in output assert "\\n" in out def test_string_with_tab_escaped(self): out = _format_toml_value("col1\tcol2") assert "\t" not in out assert "\\t" in out def test_string_with_other_controls_escaped(self): for raw, expected in [ ("\r", "\\r"), ("\f", "\\f"), ("\b", "\\b"), ]: out = _format_toml_value(f"x{raw}y") assert raw not in out, f"{raw!r} should be escaped" assert expected in out, f"{expected!r} should be in output" def test_windows_path_escaped_correctly(self): out = _format_toml_value(r"C:\Users\Alice\.codex") # Each backslash should be doubled assert out == r'"C:\\Users\\Alice\\.codex"' def test_atomic_write_no_temp_leak_on_success(self, tmp_path): """The atomic-write path uses tempfile.mkstemp + rename. On success the temp file should not be left behind.""" migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False, default_permission_profile=None) # config.toml should exist assert (tmp_path / "config.toml").exists() # And no .config.toml.* temp files left behind leftover = [p.name for p in tmp_path.iterdir() if p.name.startswith(".config.toml.")] assert leftover == [], f"temp file leaked after migration: {leftover}" def test_atomic_write_cleanup_on_rename_failure(self, tmp_path, monkeypatch): """If rename fails partway through (out of disk, permissions, crash), the temp file must be cleaned up. Otherwise repeated failed migrations would pile up .config.toml.* files.""" from pathlib import Path as _Path original_replace = _Path.replace def failing_replace(self, target): raise OSError("simulated disk full") monkeypatch.setattr(_Path, "replace", failing_replace) report = migrate( {"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False, default_permission_profile=None, ) # Error surfaced assert any("simulated disk full" in e for e in report.errors) # And no leaked temp file leftover = [p.name for p in tmp_path.iterdir() if p.name.startswith(".config.toml.")] assert leftover == [], f"temp files leaked: {leftover}" def test_unsupported_type_raises(self): with pytest.raises(ValueError): _format_toml_value(object()) class TestRenderToml: def test_starts_with_marker(self): out = render_codex_toml_section({}) assert out.startswith(MIGRATION_MARKER) def test_empty_servers_emits_placeholder(self): out = render_codex_toml_section({}) assert "no MCP servers" in out def test_servers_sorted_alphabetically(self): out = render_codex_toml_section({ "zoo": {"command": "z"}, "alpha": {"command": "a"}, "middle": {"command": "m"}, }) # Find the section header positions and confirm order a_pos = out.find("[mcp_servers.alpha]") m_pos = out.find("[mcp_servers.middle]") z_pos = out.find("[mcp_servers.zoo]") assert 0 < a_pos < m_pos < z_pos def test_server_with_args_and_env(self): out = render_codex_toml_section({ "fs": { "command": "npx", "args": ["-y", "filesystem"], "env": {"PATH": "/usr/bin"}, } }) assert "[mcp_servers.fs]" in out assert 'command = "npx"' in out assert 'args = ["-y", "filesystem"]' in out # Env emitted as inline table assert 'env = { PATH = "/usr/bin" }' in out # ---- existing-block stripping ---- class TestStripExistingManagedBlock: def test_no_managed_block_unchanged(self): text = "[other]\nfoo = 1\n" assert _strip_existing_managed_block(text) == text def test_strips_managed_block_alone(self): text = ( f"{MIGRATION_MARKER}\n" "\n" "[mcp_servers.fs]\n" 'command = "npx"\n' ) assert _strip_existing_managed_block(text).strip() == "" def test_preserves_user_content_above_managed_block(self): text = ( "[model]\n" 'name = "gpt-5.5"\n' "\n" f"{MIGRATION_MARKER}\n" "[mcp_servers.fs]\n" 'command = "x"\n' ) out = _strip_existing_managed_block(text) assert "[model]" in out assert 'name = "gpt-5.5"' in out assert "mcp_servers.fs" not in out def test_preserves_unrelated_section_after_managed_block(self): text = ( f"{MIGRATION_MARKER}\n" "[mcp_servers.fs]\n" 'command = "x"\n' "\n" "[providers]\n" 'foo = "bar"\n' ) out = _strip_existing_managed_block(text) assert "mcp_servers.fs" not in out assert "[providers]" in out assert 'foo = "bar"' in out # ---- end-to-end migrate(, expose_hermes_tools=False) ---- class TestMigrate: def test_no_servers_no_plugins_no_perms_writes_placeholder(self, tmp_path): report = migrate({}, codex_home=tmp_path, discover_plugins=False, default_permission_profile=None, expose_hermes_tools=False) assert report.written text = (tmp_path / "config.toml").read_text() assert MIGRATION_MARKER in text assert "no MCP servers" in text or "no MCP servers, plugins, or permissions" in text def test_no_servers_still_writes_permissions_default(self, tmp_path): """Even with zero MCP servers, enabling the runtime should write the default permissions profile so users don't get prompted on every write attempt. This is the fix for quirk #2.""" report = migrate({}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False) assert report.written text = (tmp_path / "config.toml").read_text() # Codex's schema: top-level `default_permissions` keying a built-in # profile name (prefixed with ":"). NOT a [permissions] section # (which is for *user-defined* profiles with structured fields). assert 'default_permissions = ":workspace"' in text assert report.wrote_permissions_default == ":workspace" def test_explicit_none_permissions_skips_block(self, tmp_path): report = migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, discover_plugins=False, default_permission_profile=None, expose_hermes_tools=False) text = (tmp_path / "config.toml").read_text() assert "default_permissions" not in text assert "[permissions]" not in text assert report.wrote_permissions_default is None def test_plugin_discovery_writes_plugin_blocks(self, tmp_path, monkeypatch): """Discovered curated plugins land as [plugins."@"] blocks. This is what OpenClaw calls 'migrate native codex plugins.'""" from hermes_cli import codex_runtime_plugin_migration as crpm def fake_query(codex_home=None, timeout=8.0): return [ {"name": "google-calendar", "marketplace": "openai-curated", "enabled": True}, {"name": "github", "marketplace": "openai-curated", "enabled": True}, ], None monkeypatch.setattr(crpm, "_query_codex_plugins", fake_query) report = migrate({}, codex_home=tmp_path, discover_plugins=True) text = (tmp_path / "config.toml").read_text() assert '[plugins."github@openai-curated"]' in text assert '[plugins."google-calendar@openai-curated"]' in text assert "enabled = true" in text assert "google-calendar@openai-curated" in report.migrated_plugins assert "github@openai-curated" in report.migrated_plugins def test_plugin_discovery_skips_unavailable_plugins(self): """Plugins where codex reports availability != AVAILABLE should be skipped — they're broken/uninstallable on codex's side, so migrating them would write config that fails at activation time. Cf. openclaw#80815.""" from hermes_cli.codex_runtime_plugin_migration import _query_codex_plugins from unittest.mock import patch # Fake a plugin/list response where one plugin is unavailable fake_response = { "marketplaces": [{ "name": "openai-curated", "plugins": [ {"name": "good-plugin", "installed": True, "enabled": True, "availability": "AVAILABLE"}, {"name": "broken-plugin", "installed": True, "enabled": True, "availability": "UNAVAILABLE"}, {"name": "auth-pending", "installed": True, "enabled": True, "availability": "REQUIRES_AUTH"}, # Plugin without availability field — pass through # (older codex versions or marketplaces that don't # set it should still work). {"name": "legacy-plugin", "installed": True, "enabled": True}, ] }] } class FakeClient: def __init__(self, **kw): pass def initialize(self, **kw): pass def request(self, method, params, timeout=None): return fake_response def close(self): pass def __enter__(self): return self def __exit__(self, *a): pass with patch("agent.transports.codex_app_server.CodexAppServerClient", FakeClient): plugins, err = _query_codex_plugins() assert err is None names = [p["name"] for p in plugins] assert "good-plugin" in names assert "legacy-plugin" in names # no field → don't skip assert "broken-plugin" not in names assert "auth-pending" not in names def test_plugin_discovery_failure_non_fatal(self, tmp_path, monkeypatch): """If codex isn't installed or RPC fails, MCP migration still completes. The error surfaces in the report but doesn't abort.""" from hermes_cli import codex_runtime_plugin_migration as crpm def fake_query_fails(codex_home=None, timeout=8.0): return [], "codex CLI not available" monkeypatch.setattr(crpm, "_query_codex_plugins", fake_query_fails) report = migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, discover_plugins=True, expose_hermes_tools=False) assert report.written assert report.migrated == ["x"] assert report.plugin_query_error == "codex CLI not available" assert report.migrated_plugins == [] def test_discover_plugins_false_skips_query(self, tmp_path, monkeypatch): """Tests and restricted environments can opt out of the subprocess spawn entirely.""" from hermes_cli import codex_runtime_plugin_migration as crpm called = {"yes": False} def boom(*a, **kw): called["yes"] = True return [], None monkeypatch.setattr(crpm, "_query_codex_plugins", boom) migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False) assert called["yes"] is False def test_dry_run_skips_plugin_query(self, tmp_path, monkeypatch): """Dry run should never spawn codex. Even with discover_plugins=True the query is skipped because dry_run takes precedence.""" from hermes_cli import codex_runtime_plugin_migration as crpm called = {"yes": False} def boom(*a, **kw): called["yes"] = True return [], None monkeypatch.setattr(crpm, "_query_codex_plugins", boom) migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, dry_run=True, discover_plugins=True, expose_hermes_tools=False) assert called["yes"] is False def test_re_run_replaces_plugin_block(self, tmp_path, monkeypatch): """Plugin blocks are managed and re-runs should replace them cleanly — same idempotency contract as MCP servers.""" from hermes_cli import codex_runtime_plugin_migration as crpm # First run: only github monkeypatch.setattr(crpm, "_query_codex_plugins", lambda codex_home=None, timeout=8.0: ( [{"name": "github", "marketplace": "openai-curated", "enabled": True}], None, )) migrate({}, codex_home=tmp_path, discover_plugins=True, default_permission_profile=None, expose_hermes_tools=False) first = (tmp_path / "config.toml").read_text() assert "github@openai-curated" in first # Second run: only canva (github went away) monkeypatch.setattr(crpm, "_query_codex_plugins", lambda codex_home=None, timeout=8.0: ( [{"name": "canva", "marketplace": "openai-curated", "enabled": True}], None, )) migrate({}, codex_home=tmp_path, discover_plugins=True, default_permission_profile=None, expose_hermes_tools=False) second = (tmp_path / "config.toml").read_text() assert "github@openai-curated" not in second assert "canva@openai-curated" in second def test_expose_hermes_tools_writes_callback_mcp_entry(self, tmp_path): """When expose_hermes_tools=True (production default), an [mcp_servers.hermes-tools] entry is written so codex calls back into Hermes for browser/web/delegate_task/vision/memory tools. This is the fix for 'all other tools that codex doesn't provide should be useable by hermes' — quirk #7.""" report = migrate({}, codex_home=tmp_path, discover_plugins=False, default_permission_profile=None, expose_hermes_tools=True) text = (tmp_path / "config.toml").read_text() assert "[mcp_servers.hermes-tools]" in text assert "hermes_tools_mcp_server" in text # Must include startup + tool timeouts so codex doesn't give up assert "startup_timeout_sec" in text assert "tool_timeout_sec" in text # And the entry is reported assert "hermes-tools" in report.migrated def test_expose_hermes_tools_disabled_skips_entry(self, tmp_path): """expose_hermes_tools=False suppresses the callback registration.""" migrate({}, codex_home=tmp_path, discover_plugins=False, default_permission_profile=None, expose_hermes_tools=False) text = (tmp_path / "config.toml").read_text() assert "[mcp_servers.hermes-tools]" not in text assert "hermes_tools_mcp_server" not in text def test_dry_run_doesnt_write(self, tmp_path): report = migrate({"mcp_servers": {"x": {"command": "y"}}}, codex_home=tmp_path, dry_run=True, expose_hermes_tools=False) assert report.dry_run is True assert not (tmp_path / "config.toml").exists() assert "x" in report.migrated def test_full_migration_round_trip(self, tmp_path): hermes_cfg = { "mcp_servers": { "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem"], }, "github": { "url": "https://api.github.com/mcp", "headers": {"Authorization": "Bearer x"}, }, } } report = migrate(hermes_cfg, codex_home=tmp_path, expose_hermes_tools=False) assert report.written text = (tmp_path / "config.toml").read_text() assert "[mcp_servers.filesystem]" in text assert "[mcp_servers.github]" in text assert 'command = "npx"' in text assert 'url = "https://api.github.com/mcp"' in text def test_idempotent_re_run_replaces_managed_block(self, tmp_path): # First migration migrate({"mcp_servers": {"a": {"command": "x"}}}, codex_home=tmp_path, expose_hermes_tools=False) first_text = (tmp_path / "config.toml").read_text() assert "[mcp_servers.a]" in first_text # Second migration with different servers migrate({"mcp_servers": {"b": {"command": "y"}}}, codex_home=tmp_path, expose_hermes_tools=False) second_text = (tmp_path / "config.toml").read_text() assert "[mcp_servers.a]" not in second_text assert "[mcp_servers.b]" in second_text def test_preserves_user_codex_config_above_marker(self, tmp_path): target = tmp_path / "config.toml" target.write_text( "[model]\n" 'profile = "default"\n' "\n" "[providers.openai]\n" 'api_key = "sk-test"\n' ) migrate({"mcp_servers": {"a": {"command": "x"}}}, codex_home=tmp_path, expose_hermes_tools=False) new_text = target.read_text() # User's codex config preserved assert "[model]" in new_text assert 'profile = "default"' in new_text assert "[providers.openai]" in new_text # And new MCP block inserted without breaking user tables assert "[mcp_servers.a]" in new_text assert MIGRATION_MARKER in new_text def test_managed_root_keys_stay_top_level_when_config_ends_in_table(self, tmp_path): """TOML has no explicit 'leave current table' syntax. If Hermes appends root keys like default_permissions after a user table such as [features], Codex parses them as features.default_permissions and rejects the config. The managed block must therefore be inserted before the first table.""" import tomllib target = tmp_path / "config.toml" target.write_text( 'model = "gpt-5.5"\n' "\n" "[features]\n" "terminal_resize_reflow = true\n" ) migrate({}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False) new_text = target.read_text() parsed = tomllib.loads(new_text) assert parsed["default_permissions"] == ":workspace" assert "default_permissions" not in parsed["features"] assert new_text.index(MIGRATION_MARKER) < new_text.index("[features]") def test_preserves_user_mcp_server_outside_managed_block(self, tmp_path): """Quirk #6: when a user adds their own MCP server entry directly to ~/.codex/config.toml outside Hermes' managed block, re-running migration must preserve it. Tested both above and below the managed block.""" target = tmp_path / "config.toml" target.write_text( "[mcp_servers.user-above]\n" 'command = "/usr/bin/above-server"\n' 'args = ["--above"]\n' ) # First migrate — adds managed block below user content migrate({"mcp_servers": {"hermes-mcp": {"command": "npx"}}}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False) text = target.read_text() assert "user-above" in text, "user MCP server above managed block got nuked" assert 'command = "/usr/bin/above-server"' in text # Append another user entry below the managed block target.write_text( text + "\n[mcp_servers.user-below]\ncommand = \"below-server\"\n" ) # Re-migrate — both should survive migrate({"mcp_servers": {"hermes-mcp": {"command": "npx"}}}, codex_home=tmp_path, discover_plugins=False, expose_hermes_tools=False) final = target.read_text() assert "user-above" in final assert "user-below" in final # And our managed block is still there with the new content assert "[mcp_servers.hermes-mcp]" in final def test_skipped_keys_reported(self, tmp_path): report = migrate({ "mcp_servers": { "x": { "command": "y", "sampling": {"enabled": True}, # codex has no equivalent } } }, codex_home=tmp_path, expose_hermes_tools=False) assert "x" in report.skipped_keys_per_server assert any("sampling" in s for s in report.skipped_keys_per_server["x"]) def test_invalid_mcp_servers_value(self, tmp_path): report = migrate({"mcp_servers": "notadict"}, codex_home=tmp_path, expose_hermes_tools=False) assert any("not a dict" in e for e in report.errors) def test_server_without_transport_skipped_with_error(self, tmp_path): report = migrate({ "mcp_servers": {"broken": {"description": "no command/url"}} }, codex_home=tmp_path, expose_hermes_tools=False) assert "broken" not in report.migrated assert any("broken" in e for e in report.errors) def test_summary_reports_migration_count(self, tmp_path): report = migrate({ "mcp_servers": {"a": {"command": "x"}, "b": {"command": "y"}} }, codex_home=tmp_path, expose_hermes_tools=False) summary = report.summary() assert "Migrated 2 MCP server(s)" in summary assert "- a" in summary assert "- b" in summary # ---- Bug B: duplicate [plugins.X] tables ---- class TestStripUnmanagedPluginTables: """Regression tests for issue #26250 Bug B. When codex itself writes ``[plugins."@"]`` tables (via the user running ``codex plugins enable`` directly), re-running ``hermes codex-runtime migrate`` would re-emit them inside the managed block and the resulting duplicate-table-header would crash codex. """ def test_strips_plugin_tables_outside_managed_block(self): text = ( 'model = "gpt-5.5"\n' "\n" "[mcp_servers.user-thing]\n" 'command = "x"\n' "\n" '[plugins."tasks@openai-curated"]\n' "enabled = true\n" "\n" '[plugins."web-search@openai-curated"]\n' "enabled = true\n" "\n" "[features]\n" "terminal_resize_reflow = true\n" ) stripped = _strip_unmanaged_plugin_tables(text) assert "[plugins." not in stripped # Non-plugin content preserved assert "[mcp_servers.user-thing]" in stripped assert "[features]" in stripped assert "terminal_resize_reflow = true" in stripped def test_preserves_content_when_no_plugin_tables(self): text = ( 'model = "gpt-5.5"\n' "\n" "[mcp_servers.x]\n" 'command = "y"\n' ) assert _strip_unmanaged_plugin_tables(text) == text def test_multi_line_array_in_plugin_table_does_not_leak(self): """A multi-line TOML array inside a [plugins.X] table whose continuation lines start with ``[`` (e.g. nested arrays) must NOT prematurely exit the strip region — otherwise array fragments leak into top-level output and produce invalid TOML on the next codex startup. Regression guard for #26260 review. """ text = ( '[plugins."tasks@openai-curated"]\n' "allowed = [\n" ' "a",\n' ' ["nested"],\n' "]\n" "[features]\n" "x = 1\n" ) stripped = _strip_unmanaged_plugin_tables(text) # Everything inside the plugin table — including the multi-line # array's continuation lines starting with `[` — should be gone. assert '["nested"]' not in stripped assert "allowed" not in stripped # Sibling user table survives intact. assert "[features]" in stripped assert "x = 1" in stripped # Result is still valid TOML. import tomllib tomllib.loads(stripped) def test_migrate_dedups_codex_owned_plugin_tables(self, tmp_path, monkeypatch): """End-to-end: codex's pre-existing [plugins.X] tables get replaced by the managed block's re-emission rather than duplicated.""" target = tmp_path / "config.toml" target.write_text( "[mcp_servers.user-server]\n" 'command = "x"\n' "\n" '[plugins."tasks@openai-curated"]\n' "enabled = true\n" ) # Simulate codex's plugin/list reporting the same plugin tasks@openai-curated. def fake_query(codex_home=None, timeout=8.0): return ( [{"name": "tasks", "marketplace": "openai-curated", "enabled": True}], None, ) monkeypatch.setattr( "hermes_cli.codex_runtime_plugin_migration._query_codex_plugins", fake_query, ) migrate({}, codex_home=tmp_path, discover_plugins=True, expose_hermes_tools=False) new_text = target.read_text() # Only ONE [plugins."tasks@openai-curated"] header should remain — inside # the managed block — not the original outside-the-block copy. assert new_text.count('[plugins."tasks@openai-curated"]') == 1 # And the surviving one is inside our managed section. managed_start = new_text.index(MIGRATION_MARKER) managed_end = new_text.index(MIGRATION_END_MARKER) plugin_idx = new_text.index('[plugins."tasks@openai-curated"]') assert managed_start < plugin_idx < managed_end # File parses cleanly as TOML (the original duplicate-key error is gone). import tomllib tomllib.loads(new_text) def test_migrate_preserves_plugin_tables_when_plugin_list_fails(self, tmp_path, monkeypatch): """If plugin/list RPC fails, we can't re-emit plugins authoritatively, so we must NOT strip the user's existing [plugins.X] tables — that would silently lose them.""" target = tmp_path / "config.toml" target.write_text( '[plugins."tasks@openai-curated"]\n' "enabled = true\n" ) def fake_query(codex_home=None, timeout=8.0): return ([], "plugin/list query failed: codex not installed") monkeypatch.setattr( "hermes_cli.codex_runtime_plugin_migration._query_codex_plugins", fake_query, ) migrate({}, codex_home=tmp_path, discover_plugins=True, expose_hermes_tools=False) new_text = target.read_text() # User's plugin table preserved verbatim — we can't re-emit it. assert '[plugins."tasks@openai-curated"]' in new_text # ---- Bug C: HERMES_HOME tempdir leak into ~/.codex/config.toml ---- class TestHermesHomeLeakGuard: """Regression tests for issue #26250 Bug C. Previously ``_build_hermes_tools_mcp_entry()`` read ``HERMES_HOME`` directly from ``os.environ``, so a pytest ``monkeypatch.setenv`` would leak a transient tempdir path into the user's real ``~/.codex/config.toml`` once codex spawned the hermes-tools MCP subprocess. """ def test_tempdir_detector_recognizes_pytest_paths(self): assert _looks_like_test_tempdir( "/private/var/folders/abc/pytest-of-kshitij/pytest-137/popen-gw2/test_X/hermes_test" ) assert _looks_like_test_tempdir( "/tmp/pytest-of-user/pytest-12/test_X/hermes" ) assert _looks_like_test_tempdir( "/private/var/folders/zz/T/pytest-of-bob/pytest-1" ) def test_tempdir_detector_accepts_real_hermes_home(self): assert not _looks_like_test_tempdir("/Users/alice/.hermes") assert not _looks_like_test_tempdir("/home/bob/.hermes") assert not _looks_like_test_tempdir("/opt/hermes") assert not _looks_like_test_tempdir("") def test_pytest_tempdir_not_burned_into_mcp_env(self, monkeypatch): """The headline regression: even when HERMES_HOME points at a pytest tempdir, _build_hermes_tools_mcp_entry() must NOT propagate it.""" monkeypatch.setenv( "HERMES_HOME", "/private/var/folders/xx/pytest-of-user/pytest-99/test_x/hermes_test", ) entry = _build_hermes_tools_mcp_entry() env = entry.get("env", {}) assert "HERMES_HOME" not in env, ( f"pytest-tempdir HERMES_HOME leaked into codex MCP entry: " f"{env.get('HERMES_HOME')!r}" ) def test_real_hermes_home_propagates(self, monkeypatch, tmp_path): """A legitimate HERMES_HOME (not a tempdir path) DOES propagate so the MCP subprocess sees the same config as the parent CLI.""" # Use a path that looks real — under /Users or /home, not /var/folders. # We can't easily create one in the test, so just use a stable path # outside any tempdir-detector needle. The detector checks for tempdir # markers, not for path existence. real_path = "/Users/alice/.hermes" monkeypatch.setenv("HERMES_HOME", real_path) entry = _build_hermes_tools_mcp_entry() env = entry.get("env", {}) assert env.get("HERMES_HOME") == real_path def test_unset_hermes_home_omits_env_key(self, monkeypatch): """When HERMES_HOME is unset in the environment, the MCP entry MUST NOT bake in a resolved-default path. The codex subprocess should inherit whatever HERMES_HOME its launcher (systemd, gateway, shell) sets at runtime, rather than being pinned to migrate-time defaults. Regression guard for issue #26250 follow-up review.""" monkeypatch.delenv("HERMES_HOME", raising=False) entry = _build_hermes_tools_mcp_entry() env = entry.get("env", {}) assert "HERMES_HOME" not in env, ( f"HERMES_HOME should not be set when env var is unset, got: " f"{env.get('HERMES_HOME')!r}" )