"""Tests for ``muse agent-config``. Coverage -------- Unit _detect_context — standalone, workspace_root, workspace_member _compute_rel_path — relative path from repo to workspace root _render_adapter — include syntax (Claude) vs embedded (Codex/Cursor/etc.) _load_configured_adapters — reads [agent-config] adapters from config.toml Integration — init standalone — generates .muse/agent.md with full content workspace_root — generates workspace-level .muse/agent.md with member table workspace_member — generates thin repo-level .muse/agent.md linking to workspace --force — overwrites existing .muse/agent.md no --force on existing — exits 1 with clear message --json schema — all fields present Integration — sync standalone — generates all adapter files from .muse/agent.md workspace_member — Claude adapter includes both workspace + repo level embed adapters — non-Claude adapters embed full content --adapters claude — only generates CLAUDE.md --dry-run — prints what would be written, no files created --force — overwrites existing adapter files no --force on existing — exits 1 missing agent.md — exits 1 with helpful message --json schema — all fields present config.toml adapters — sync respects [agent-config] adapters setting --adapters overrides — CLI flag takes priority over config.toml Integration — show standalone — prints .muse/agent.md content merged workspace — prints workspace + repo content concatenated --json — content field present Integration — status all adapters present — reports in_sync correctly some missing — reports missing --json schema — all fields present Integration — set writes config.toml — [agent-config] adapters persisted correctly updates existing — existing [agent-config] section replaced preserves other keys — other config.toml sections untouched unknown adapter exits1 — invalid adapter name exits with code 1 --json schema — {adapters, path} present E2E — full workflow init → set → sync → edit → status out-of-sync → sync --force → status clean Stress large agent.md — 200 KB content syncs without error rapid sequential syncs — 30 iterations stable Data Integrity sync is atomic — adapter file is never partially written corrupt config.toml — falls back to all adapters gracefully Performance sync completes quickly — wall time < 2 s Security set rejects path traversal in adapter name malformed config.toml — TOML injection attempt doesn't crash agent.md with null bytes — handled without crash """ from __future__ import annotations import json import pathlib import time import threading import pytest from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _invoke(path: pathlib.Path, args: list[str]) -> object: import os saved = os.getcwd() try: os.chdir(path) return runner.invoke(None, args) finally: os.chdir(saved) def _init_repo(path: pathlib.Path, domain: str = "code") -> None: r = _invoke(path, ["init", "--domain", domain]) assert r.exit_code == 0, r.output def _init_workspace(path: pathlib.Path, members: list[tuple[str, str]]) -> None: """Create a workspace manifest at path with given (name, rel_path) members.""" muse_dir = path / ".muse" muse_dir.mkdir(parents=True, exist_ok=True) lines = [""] for name, rel in members: lines += [ "[[members]]", f'name = "{name}"', f'url = "http://localhost:10003/gabriel/{name}"', f'path = "{rel}"', 'branch = "main"', "", ] (muse_dir / "workspace.toml").write_text("\n".join(lines)) # --------------------------------------------------------------------------- # Unit — _detect_context # --------------------------------------------------------------------------- class TestDetectContext: def test_standalone_repo(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _detect_context _init_repo(tmp_path) kind, ws = _detect_context(tmp_path) assert kind == "standalone" assert ws is None def test_workspace_root(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _detect_context _init_workspace(tmp_path, [("muse", "muse")]) kind, ws = _detect_context(tmp_path) assert kind == "workspace_root" assert ws == tmp_path def test_workspace_member(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _detect_context _init_workspace(tmp_path, [("core", "core")]) repo = tmp_path / "core" repo.mkdir() _init_repo(repo) kind, ws = _detect_context(repo) assert kind == "workspace_member" assert ws == tmp_path # --------------------------------------------------------------------------- # Unit — _compute_rel_path # --------------------------------------------------------------------------- class TestComputeRelPath: def test_direct_child(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _compute_rel_path ws = tmp_path / "ws" repo = tmp_path / "ws" / "core" ws.mkdir(), repo.mkdir() assert _compute_rel_path(repo, ws) == ".." def test_nested_child(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _compute_rel_path ws = tmp_path / "ws" repo = tmp_path / "ws" / "packages" / "foo" repo.mkdir(parents=True) assert _compute_rel_path(repo, ws) == "../.." def test_same_dir(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _compute_rel_path assert _compute_rel_path(tmp_path, tmp_path) == "." # --------------------------------------------------------------------------- # Unit — _render_adapter # --------------------------------------------------------------------------- class TestRenderAdapter: def test_include_adapter_uses_at_syntax(self) -> None: from muse.cli.commands.agent_config import _render_adapter, _ADAPTERS spec = _ADAPTERS["claude"] result = _render_adapter(spec, repo_agent_md=".muse/agent.md", ws_agent_md=None) assert "@.muse/agent.md" in result assert "embed" not in result.lower() def test_include_adapter_with_workspace(self) -> None: from muse.cli.commands.agent_config import _render_adapter, _ADAPTERS spec = _ADAPTERS["claude"] result = _render_adapter(spec, repo_agent_md=".muse/agent.md", ws_agent_md="../.muse/agent.md") assert "@../.muse/agent.md" in result assert "@.muse/agent.md" in result def test_embed_adapter_contains_content(self) -> None: from muse.cli.commands.agent_config import _render_adapter, _ADAPTERS spec = _ADAPTERS["codex"] result = _render_adapter( spec, repo_agent_md=".muse/agent.md", ws_agent_md=None, repo_agent_content="# My Agent Config\nsome rules", ws_agent_content=None, ) assert "# My Agent Config" in result assert "some rules" in result def test_embed_adapter_with_workspace_prepends_ws_content(self) -> None: from muse.cli.commands.agent_config import _render_adapter, _ADAPTERS spec = _ADAPTERS["codex"] result = _render_adapter( spec, repo_agent_md=".muse/agent.md", ws_agent_md="../.muse/agent.md", repo_agent_content="# Repo Config", ws_agent_content="# Workspace Config", ) ws_pos = result.index("# Workspace Config") repo_pos = result.index("# Repo Config") assert ws_pos < repo_pos # workspace content comes first # --------------------------------------------------------------------------- # Integration — init: standalone repo # --------------------------------------------------------------------------- class TestInitStandalone: def test_creates_agent_md(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "init"]) assert result.exit_code == 0 assert (tmp_path / ".muse" / "agent.md").exists() def test_agent_md_contains_muse_rule(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert "Muse" in content assert "git" in content.lower() # the no-git rule mentions "git" def test_agent_md_contains_branch_flow(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert "checkout -b" in content def test_agent_md_contains_repo_name(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert tmp_path.name in content def test_no_force_on_existing_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "init"]) assert result.exit_code == 1 assert "force" in result.output.lower() or "--force" in result.output def test_force_overwrites(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) (tmp_path / ".muse" / "agent.md").write_text("old content") _invoke(tmp_path, ["agent-config", "init", "--force"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert content != "old content" assert "Muse" in content def test_json_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "init", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "path" in data assert "scope" in data assert "created" in data # --------------------------------------------------------------------------- # Integration — init: workspace root # --------------------------------------------------------------------------- class TestInitWorkspaceRoot: def test_creates_agent_md_at_workspace_root(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core"), ("api", "api")]) result = _invoke(tmp_path, ["agent-config", "init"]) assert result.exit_code == 0 assert (tmp_path / ".muse" / "agent.md").exists() def test_workspace_agent_md_lists_members(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core"), ("api", "api")]) _invoke(tmp_path, ["agent-config", "init"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert "core" in content assert "api" in content def test_workspace_agent_md_contains_shared_rules(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core")]) _invoke(tmp_path, ["agent-config", "init"]) content = (tmp_path / ".muse" / "agent.md").read_text() assert "Muse" in content assert "git" in content.lower() # --------------------------------------------------------------------------- # Integration — init: workspace member # --------------------------------------------------------------------------- class TestInitWorkspaceMember: def test_creates_repo_level_agent_md(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core")]) repo = tmp_path / "core" repo.mkdir() _init_repo(repo) result = _invoke(repo, ["agent-config", "init"]) assert result.exit_code == 0 assert (repo / ".muse" / "agent.md").exists() def test_member_agent_md_references_workspace(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core")]) repo = tmp_path / "core" repo.mkdir() _init_repo(repo) _invoke(repo, ["agent-config", "init"]) content = (repo / ".muse" / "agent.md").read_text() # Should mention the workspace or link to the parent config assert "workspace" in content.lower() or ".muse/agent.md" in content # --------------------------------------------------------------------------- # Integration — sync # --------------------------------------------------------------------------- class TestSync: @pytest.fixture() def standalone(self, tmp_path: pathlib.Path) -> pathlib.Path: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) return tmp_path def test_sync_creates_claude_md(self, standalone: pathlib.Path) -> None: result = _invoke(standalone, ["agent-config", "sync"]) assert result.exit_code == 0 assert (standalone / "CLAUDE.md").exists() def test_sync_creates_agents_md(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) assert (standalone / "AGENTS.md").exists() def test_sync_creates_cursorrules(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) assert (standalone / ".cursorrules").exists() def test_sync_creates_windsurfrules(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) assert (standalone / ".windsurfrules").exists() def test_claude_md_uses_include_syntax(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) content = (standalone / "CLAUDE.md").read_text() assert "@.muse/agent.md" in content def test_agents_md_embeds_content(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) agent_md_content = (standalone / ".muse" / "agent.md").read_text() agents_md_content = (standalone / "AGENTS.md").read_text() # Should contain actual text, not an @ include assert "@" not in agents_md_content.split("\n")[2] # not just an include # Should contain meaningful content from agent.md assert "Muse" in agents_md_content def test_sync_adapters_flag_limits_output(self, standalone: pathlib.Path) -> None: result = _invoke(standalone, ["agent-config", "sync", "--adapters", "claude"]) assert result.exit_code == 0 assert (standalone / "CLAUDE.md").exists() assert not (standalone / "AGENTS.md").exists() def test_dry_run_creates_no_files(self, standalone: pathlib.Path) -> None: result = _invoke(standalone, ["agent-config", "sync", "--dry-run"]) assert result.exit_code == 0 assert not (standalone / "CLAUDE.md").exists() assert not (standalone / "AGENTS.md").exists() def test_dry_run_prints_what_would_be_written(self, standalone: pathlib.Path) -> None: result = _invoke(standalone, ["agent-config", "sync", "--dry-run"]) assert "CLAUDE.md" in result.output or "claude" in result.output.lower() def test_no_force_on_existing_exits_1(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) result = _invoke(standalone, ["agent-config", "sync"]) assert result.exit_code == 1 def test_force_overwrites_existing(self, standalone: pathlib.Path) -> None: _invoke(standalone, ["agent-config", "sync"]) (standalone / "CLAUDE.md").write_text("old content") result = _invoke(standalone, ["agent-config", "sync", "--force"]) assert result.exit_code == 0 content = (standalone / "CLAUDE.md").read_text() assert content != "old content" def test_missing_agent_md_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 1 assert "agent.md" in result.output.lower() or "init" in result.output.lower() def test_json_schema(self, standalone: pathlib.Path) -> None: result = _invoke(standalone, ["agent-config", "sync", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "adapters" in data assert isinstance(data["adapters"], list) for entry in data["adapters"]: assert "name" in entry assert "path" in entry assert "written" in entry def test_workspace_member_claude_includes_both_levels( self, tmp_path: pathlib.Path ) -> None: _init_workspace(tmp_path, [("core", "core")]) # Init workspace-level agent.md _invoke(tmp_path, ["agent-config", "init"]) # Init and sync repo-level repo = tmp_path / "core" repo.mkdir() _init_repo(repo) _invoke(repo, ["agent-config", "init"]) _invoke(repo, ["agent-config", "sync"]) content = (repo / "CLAUDE.md").read_text() # Should include both workspace level and repo level assert "agent.md" in content # Workspace-level reference should be present (parent path) assert ".." in content # --------------------------------------------------------------------------- # Integration — show # --------------------------------------------------------------------------- class TestShow: def test_show_prints_agent_md_content(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "show"]) assert result.exit_code == 0 agent_md = (tmp_path / ".muse" / "agent.md").read_text() assert agent_md.strip() in result.output def test_show_missing_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "show"]) assert result.exit_code == 1 def test_show_json_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "show", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "content" in data assert "path" in data assert "scope" in data def test_show_merged_workspace(self, tmp_path: pathlib.Path) -> None: _init_workspace(tmp_path, [("core", "core")]) _invoke(tmp_path, ["agent-config", "init"]) repo = tmp_path / "core" repo.mkdir() _init_repo(repo) _invoke(repo, ["agent-config", "init"]) result = _invoke(repo, ["agent-config", "show", "--scope", "merged"]) assert result.exit_code == 0 # Should include content from both levels ws_content = (tmp_path / ".muse" / "agent.md").read_text() repo_content = (repo / ".muse" / "agent.md").read_text() assert ws_content[:30] in result.output or repo_content[:30] in result.output # --------------------------------------------------------------------------- # Integration — status # --------------------------------------------------------------------------- class TestStatus: def test_status_before_sync(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "status"]) assert result.exit_code == 0 # All adapters should show as missing assert "CLAUDE.md" in result.output or "claude" in result.output.lower() def test_status_after_sync(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "status"]) assert result.exit_code == 0 def test_status_json_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "status", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "agent_md" in data assert "adapters" in data for entry in data["adapters"]: assert "name" in entry assert "filename" in entry assert "exists" in entry assert "in_sync" in entry def test_status_shows_out_of_sync_after_edit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) # Modify agent.md without re-syncing agent_md = tmp_path / ".muse" / "agent.md" agent_md.write_text(agent_md.read_text() + "\n# NEW RULE\n") result = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(result.output) # At least one embed adapter should be out of sync embed_adapters = [a for a in data["adapters"] if a["name"] != "claude"] assert any(not a["in_sync"] for a in embed_adapters) # --------------------------------------------------------------------------- # Unit — _load_configured_adapters # --------------------------------------------------------------------------- class TestLoadConfiguredAdapters: def test_returns_none_when_no_config_toml(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) assert _load_configured_adapters(tmp_path) is None def test_returns_none_when_no_agent_config_section(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text('[hub]\nurl = "http://localhost:10003"\n') assert _load_configured_adapters(tmp_path) is None def test_returns_list_when_set(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text('[agent-config]\nadapters = ["claude", "codex"]\n') assert _load_configured_adapters(tmp_path) == ["claude", "codex"] def test_returns_none_for_malformed_list(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text('[agent-config]\nadapters = "not-a-list"\n') assert _load_configured_adapters(tmp_path) is None def test_returns_none_for_corrupt_toml(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text("[[[[invalid toml") assert _load_configured_adapters(tmp_path) is None # --------------------------------------------------------------------------- # Integration — set subcommand # --------------------------------------------------------------------------- class TestSet: def test_writes_config_toml(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex"]) assert result.exit_code == 0 config = (tmp_path / ".muse" / "config.toml").read_text() assert "claude" in config assert "codex" in config def test_json_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "adapters" in data assert "path" in data assert data["adapters"] == ["claude"] def test_updates_existing_section(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex"]) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude"]) config = (tmp_path / ".muse" / "config.toml").read_text() # Only one [agent-config] section assert config.count("[agent-config]") == 1 # codex no longer present in the adapters list import tomllib raw = tomllib.loads(config) assert raw["agent-config"]["adapters"] == ["claude"] def test_preserves_other_config_sections(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text('[hub]\nurl = "http://localhost:10003"\n') _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude"]) config = (tmp_path / ".muse" / "config.toml").read_text() assert "[hub]" in config assert "localhost:10003" in config assert "[agent-config]" in config def test_unknown_adapter_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "vscode"]) assert result.exit_code == 1 def test_unknown_adapter_error_message(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "vscode"]) assert "vscode" in result.output.lower() or "unknown" in result.output.lower() def test_single_adapter(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude", "--json"]) assert result.exit_code == 0 assert json.loads(result.output)["adapters"] == ["claude"] def test_all_adapters_accepted(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _ADAPTERS _init_repo(tmp_path) all_names = ",".join(_ADAPTERS.keys()) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", all_names]) assert result.exit_code == 0 # --------------------------------------------------------------------------- # Integration — sync priority chain # --------------------------------------------------------------------------- class TestSyncPriorityChain: def test_config_toml_limits_adapters(self, tmp_path: pathlib.Path) -> None: """[agent-config] adapters in config.toml limits sync without --adapters.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude"]) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 0 assert (tmp_path / "CLAUDE.md").exists() assert not (tmp_path / "AGENTS.md").exists() def test_cli_adapters_flag_overrides_config_toml(self, tmp_path: pathlib.Path) -> None: """--adapters on CLI takes priority over config.toml setting.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude"]) result = _invoke(tmp_path, ["agent-config", "sync", "--adapters", "codex"]) assert result.exit_code == 0 assert (tmp_path / "AGENTS.md").exists() assert not (tmp_path / "CLAUDE.md").exists() def test_no_config_generates_all_adapters(self, tmp_path: pathlib.Path) -> None: """Without config.toml setting, all adapters are generated.""" from muse.cli.commands.agent_config import _ADAPTERS _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) for spec in _ADAPTERS.values(): assert (tmp_path / spec["filename"]).exists(), f"missing {spec['filename']}" # --------------------------------------------------------------------------- # E2E — full workflow # --------------------------------------------------------------------------- class TestE2EFullWorkflow: def test_init_set_sync_edit_status_resync(self, tmp_path: pathlib.Path) -> None: """Complete agent-config lifecycle: init → set → sync → edit → out-of-sync → fix.""" _init_repo(tmp_path) # init r = _invoke(tmp_path, ["agent-config", "init"]) assert r.exit_code == 0 assert (tmp_path / ".muse" / "agent.md").exists() # set r = _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex"]) assert r.exit_code == 0 # sync r = _invoke(tmp_path, ["agent-config", "sync"]) assert r.exit_code == 0 assert (tmp_path / "CLAUDE.md").exists() assert (tmp_path / "AGENTS.md").exists() # status — in sync r = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(r.output) active = [a for a in data["adapters"] if a["exists"]] assert all(a["in_sync"] for a in active) # edit agent.md agent_md = tmp_path / ".muse" / "agent.md" agent_md.write_text(agent_md.read_text() + "\n# EXTRA RULE\n") # status — codex out of sync (embed adapter) r = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(r.output) codex = next(a for a in data["adapters"] if a["name"] == "codex") assert not codex["in_sync"] # sync --force r = _invoke(tmp_path, ["agent-config", "sync", "--force"]) assert r.exit_code == 0 # status — back in sync r = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(r.output) active = [a for a in data["adapters"] if a["exists"]] assert all(a["in_sync"] for a in active) # verify new content is in AGENTS.md assert "EXTRA RULE" in (tmp_path / "AGENTS.md").read_text() def test_workspace_e2e(self, tmp_path: pathlib.Path) -> None: """Workspace hierarchy: shared rules flow into member CLAUDE.md.""" _init_workspace(tmp_path, [("core", "core")]) _invoke(tmp_path, ["agent-config", "init"]) repo = tmp_path / "core" repo.mkdir() _init_repo(repo) _invoke(repo, ["agent-config", "init"]) _invoke(repo, ["agent-config", "sync"]) claude = (repo / "CLAUDE.md").read_text() assert "@../.muse/agent.md" in claude assert "@.muse/agent.md" in claude # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_large_agent_md_syncs(self, tmp_path: pathlib.Path) -> None: """200 KB agent.md embeds correctly into AGENTS.md.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) # Overwrite with 200 KB of content large = "# Rule\n" + ("x" * 200) + "\n" large_content = large * 1000 # ~200 KB (tmp_path / ".muse" / "agent.md").write_text(large_content) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 0 agents_md = (tmp_path / "AGENTS.md").read_text() assert len(agents_md) > 100_000 def test_rapid_sequential_syncs(self, tmp_path: pathlib.Path) -> None: """30 sequential sync --force calls produce consistent output.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) content_before = (tmp_path / "AGENTS.md").read_text() for _ in range(30): r = _invoke(tmp_path, ["agent-config", "sync", "--force"]) assert r.exit_code == 0 assert (tmp_path / "AGENTS.md").read_text() == content_before def test_concurrent_sync_no_corruption(self, tmp_path: pathlib.Path) -> None: """Concurrent sync --force calls never produce a torn file.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) expected = (tmp_path / "AGENTS.md").read_text() errors: list[str] = [] def sync() -> None: r = _invoke(tmp_path, ["agent-config", "sync", "--force"]) if r.exit_code != 0: errors.append(r.output) threads = [threading.Thread(target=sync) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors # File is never empty or partial — must be valid content result = (tmp_path / "AGENTS.md").read_text() assert len(result) > 0 assert "Muse" in result # --------------------------------------------------------------------------- # Data Integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_corrupt_config_toml_falls_back_to_all_adapters( self, tmp_path: pathlib.Path ) -> None: """Corrupt config.toml is silently ignored — all adapters generated.""" from muse.cli.commands.agent_config import _ADAPTERS _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) (tmp_path / ".muse" / "config.toml").write_text("[[[[not valid toml") result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 0 for spec in _ADAPTERS.values(): assert (tmp_path / spec["filename"]).exists() def test_adapter_file_not_empty_after_sync(self, tmp_path: pathlib.Path) -> None: """Every generated adapter file has non-zero content.""" from muse.cli.commands.agent_config import _ADAPTERS _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) for spec in _ADAPTERS.values(): p = tmp_path / spec["filename"] assert p.stat().st_size > 0, f"{spec['filename']} is empty" def test_sync_write_is_atomic(self, tmp_path: pathlib.Path) -> None: """After sync, AGENTS.md is a complete file — not truncated mid-write.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) content = (tmp_path / "AGENTS.md").read_text() # Content should end with a newline, not be truncated mid-line assert content.endswith("\n") def test_set_preserves_existing_config_integrity( self, tmp_path: pathlib.Path ) -> None: """set writes valid TOML that can be re-parsed.""" import tomllib _init_repo(tmp_path) (tmp_path / ".muse" / "config.toml").write_text( '[hub]\nurl = "http://localhost:10003"\n\n[limits]\nmax_file_size_mb = 10\n' ) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex"]) raw = tomllib.loads((tmp_path / ".muse" / "config.toml").read_text()) assert raw["hub"]["url"] == "http://localhost:10003" assert raw["limits"]["max_file_size_mb"] == 10 assert raw["agent-config"]["adapters"] == ["claude", "codex"] # --------------------------------------------------------------------------- # Performance # --------------------------------------------------------------------------- class TestPerformance: def test_sync_completes_under_2_seconds(self, tmp_path: pathlib.Path) -> None: """sync with default adapters completes in under 2 seconds.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) start = time.monotonic() _invoke(tmp_path, ["agent-config", "sync"]) elapsed = time.monotonic() - start assert elapsed < 2.0, f"sync took {elapsed:.2f}s — too slow" def test_status_completes_under_1_second(self, tmp_path: pathlib.Path) -> None: """status check completes in under 1 second.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) start = time.monotonic() _invoke(tmp_path, ["agent-config", "status", "--json"]) elapsed = time.monotonic() - start assert elapsed < 1.0, f"status took {elapsed:.2f}s — too slow" # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_set_rejects_path_traversal_in_adapter_name( self, tmp_path: pathlib.Path ) -> None: """set does not accept adapter names containing path separators.""" _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "../evil"]) assert result.exit_code == 1 def test_set_rejects_adapter_with_null_byte( self, tmp_path: pathlib.Path ) -> None: """set rejects adapter names containing null bytes.""" _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude\x00evil"]) assert result.exit_code == 1 def test_agent_md_with_null_bytes_does_not_crash_sync( self, tmp_path: pathlib.Path ) -> None: """agent.md containing null bytes is handled without an unhandled exception.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) # Write null bytes into agent.md agent_md = tmp_path / ".muse" / "agent.md" agent_md.write_bytes(agent_md.read_bytes() + b"\x00\x00malicious\x00") # Should not raise — exit code may be 0 or 1 but must not be an unhandled exception result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code in (0, 1) def test_toml_injection_in_config_does_not_escape_section( self, tmp_path: pathlib.Path ) -> None: """A crafted adapter name cannot inject extra TOML sections.""" import tomllib _init_repo(tmp_path) # Attempt to inject a new TOML section via adapter name result = _invoke( tmp_path, ["agent-config", "set", "--adapters", 'claude"]\n[injected'], ) # Should fail with unknown adapter error, not write injected TOML assert result.exit_code == 1 config_path = tmp_path / ".muse" / "config.toml" if config_path.exists(): raw = tomllib.loads(config_path.read_text()) assert "injected" not in raw