"""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 argparse import json import pathlib import time import threading import pytest from muse.core.paths import agent_md_path, config_toml_path, muse_dir from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _invoke(path: pathlib.Path, args: list[str]) -> "InvokeResult": 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_with_all_adapters(path: pathlib.Path) -> None: """init + set all adapters — use in tests that need all adapter files generated.""" _init_repo(path) _invoke(path, ["agent-config", "init"]) _invoke(path, ["agent-config", "set", "--adapters", "claude,codex,cursor,windsurf"]) def _init_workspace(path: pathlib.Path, members: list[tuple[str, str]]) -> None: """Create a workspace manifest at path with given (name, rel_path) members.""" dot_muse = muse_dir(path) dot_muse.mkdir(parents=True, exist_ok=True) lines = [""] for name, rel in members: lines += [ "[[members]]", f'name = "{name}"', f'url = "https://localhost:1337/gabriel/{name}"', f'path = "{rel}"', 'branch = "main"', "", ] (dot_muse / "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 (agent_md_path(tmp_path)).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 = (agent_md_path(tmp_path)).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 = (agent_md_path(tmp_path)).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 = (agent_md_path(tmp_path)).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.stderr.lower() or "--force" in result.stderr def test_force_overwrites(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) (agent_md_path(tmp_path)).write_text("old content") _invoke(tmp_path, ["agent-config", "init", "--force"]) content = (agent_md_path(tmp_path)).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 (agent_md_path(tmp_path)).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 = (agent_md_path(tmp_path)).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 = (agent_md_path(tmp_path)).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 (agent_md_path(repo)).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 = (agent_md_path(repo)).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"]) # Explicitly configure all adapters so tests that want specific files work. _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex,cursor,windsurf"]) return tmp_path def test_sync_requires_adapter_config(self, tmp_path: pathlib.Path) -> None: """sync with no [agent-config] section exits with error instead of writing all adapters.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code != 0 assert "agent-config set" in result.stderr or "agent-config set" in result.output 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 = (agent_md_path(standalone)).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_sync_claude_only_config_writes_only_claude(self, tmp_path: pathlib.Path) -> None: """When adapters = [claude], sync writes ONLY CLAUDE.md — nothing else.""" _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() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").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_sync_already_in_sync_skips_without_error(self, standalone: pathlib.Path) -> None: """Second sync with no changes skips in-sync files and exits 0.""" _invoke(standalone, ["agent-config", "sync"]) result = _invoke(standalone, ["agent-config", "sync"]) assert result.exit_code == 0 # Output should indicate files were skipped assert "in sync" in result.output or "skipped" in result.output.lower() or result.exit_code == 0 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.stderr.lower() or "init" in result.stderr.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_with_all_adapters(repo) _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 — read # --------------------------------------------------------------------------- class TestRead: def test_read_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", "read"]) assert result.exit_code == 0 agent_md = (agent_md_path(tmp_path)).read_text() assert agent_md.strip() in result.output def test_read_missing_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "read"]) assert result.exit_code == 1 def test_read_json_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "read", "--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_read_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", "read", "--scope", "merged"]) assert result.exit_code == 0 # Should include content from both levels ws_content = (agent_md_path(tmp_path)).read_text() repo_content = (agent_md_path(repo)).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 = agent_md_path(tmp_path) agent_md.write_text(f"{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: """All tests isolate user-level config via MUSE_USER_CONFIG_DIR so the real ~/.muse/config.toml never interferes with the expected result.""" @pytest.fixture(autouse=True) def isolate_user_config(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """Point MUSE_USER_CONFIG_DIR at a fresh tmp dir — no real user config.""" user_dir = tmp_path / "user_muse" user_dir.mkdir() monkeypatch.setenv("MUSE_USER_CONFIG_DIR", str(user_dir)) 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) (config_toml_path(tmp_path)).write_text('[hub]\nurl = "https://localhost:1337"\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) (config_toml_path(tmp_path)).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) (config_toml_path(tmp_path)).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) (config_toml_path(tmp_path)).write_text("[[[[invalid toml") assert _load_configured_adapters(tmp_path) is None def test_falls_back_to_user_config_when_no_repo_config( self, tmp_path: pathlib.Path ) -> None: """When repo has no [agent-config], user-level config is used as fallback.""" import os as _os from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) user_dir = pathlib.Path(_os.environ["MUSE_USER_CONFIG_DIR"]) (user_dir / "config.toml").write_text('[agent-config]\nadapters = ["claude"]\n') assert _load_configured_adapters(tmp_path) == ["claude"] def test_repo_config_takes_priority_over_user_config( self, tmp_path: pathlib.Path ) -> None: """Repo-level [agent-config] overrides the user-level fallback.""" import os as _os from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) user_dir = pathlib.Path(_os.environ["MUSE_USER_CONFIG_DIR"]) (user_dir / "config.toml").write_text('[agent-config]\nadapters = ["codex"]\n') (config_toml_path(tmp_path)).write_text('[agent-config]\nadapters = ["claude"]\n') # Repo says claude; user says codex — repo wins assert _load_configured_adapters(tmp_path) == ["claude"] def test_user_config_fallback_absent_returns_none( self, tmp_path: pathlib.Path ) -> None: """Both repo and user config absent → None.""" from muse.cli.commands.agent_config import _load_configured_adapters _init_repo(tmp_path) 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 = (config_toml_path(tmp_path)).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 = (config_toml_path(tmp_path)).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) (config_toml_path(tmp_path)).write_text('[hub]\nurl = "https://localhost:1337"\n') _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude"]) config = (config_toml_path(tmp_path)).read_text() assert "[hub]" in config assert "localhost:1337" 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.stderr.lower() or "unknown" in result.stderr.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 def test_global_flag_writes_to_user_config( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--global writes to MUSE_USER_CONFIG_DIR/config.toml, not the repo.""" user_dir = tmp_path / "user_muse" user_dir.mkdir() monkeypatch.setenv("MUSE_USER_CONFIG_DIR", str(user_dir)) _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "set", "--global", "--adapters", "claude"]) assert result.exit_code == 0 user_cfg = (user_dir / "config.toml").read_text() assert "claude" in user_cfg # Repo config must NOT have the section repo_cfg = config_toml_path(tmp_path) if repo_cfg.exists(): assert "[agent-config]" not in repo_cfg.read_text() def test_global_flag_survives_repo_absence( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--global works even when CWD is not inside a muse repo.""" user_dir = tmp_path / "user_muse" user_dir.mkdir() monkeypatch.setenv("MUSE_USER_CONFIG_DIR", str(user_dir)) # Use a directory with no .muse/ — repo is NOT required for --global non_repo = tmp_path / "not_a_repo" non_repo.mkdir() _init_repo(non_repo) # init so we have a valid CWD repo context result = _invoke(non_repo, ["agent-config", "set", "--global", "--adapters", "claude"]) assert result.exit_code == 0 assert "claude" in (user_dir / "config.toml").read_text() def test_global_adapters_visible_to_sync( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """sync picks up global config when the repo has no [agent-config] section.""" user_dir = tmp_path / "user_muse" user_dir.mkdir() monkeypatch.setenv("MUSE_USER_CONFIG_DIR", str(user_dir)) _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "set", "--global", "--adapters", "claude"]) # Repo has no [agent-config] — should fall back to global 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() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").exists() # --------------------------------------------------------------------------- # 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_exits_with_error(self, tmp_path: pathlib.Path) -> None: """Without [agent-config] adapters set, sync exits with an actionable error.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code != 0 assert "agent-config set" in result.stderr or "agent-config set" in result.output # --------------------------------------------------------------------------- # 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 (agent_md_path(tmp_path)).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 = agent_md_path(tmp_path) agent_md.write_text(f"{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_with_all_adapters(repo) _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_with_all_adapters(tmp_path) # Overwrite with 200 KB of content large = f"# Rule\n{'x' * 200}\n" large_content = large * 1000 # ~200 KB (agent_md_path(tmp_path)).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_with_all_adapters(tmp_path) _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. Uses write_text_atomic directly to test the atomicity guarantee without threading through the full CLI (which relies on process-global CWD). """ from muse.core.io import write_text_atomic target = tmp_path / "AGENTS.md" content = "# Agent rules\n" + "x" * 10_000 + "\n" errors: list[str] = [] def write() -> None: try: write_text_atomic(target, content) except Exception as exc: errors.append(str(exc)) threads = [threading.Thread(target=write) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors result = target.read_text() assert len(result) > 0 # File must be complete — never a partial write assert result == content # --------------------------------------------------------------------------- # Data Integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_corrupt_config_toml_exits_with_error( self, tmp_path: pathlib.Path ) -> None: """Corrupt config.toml causes sync to fail — no files are silently generated.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) (config_toml_path(tmp_path)).write_text("[[[[not valid toml") result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code != 0 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_with_all_adapters(tmp_path) _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_with_all_adapters(tmp_path) _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) (config_toml_path(tmp_path)).write_text( '[hub]\nurl = "https://localhost:1337"\n\n[limits]\nmax_file_size_mb = 10\n' ) _invoke(tmp_path, ["agent-config", "set", "--adapters", "claude,codex"]) raw = tomllib.loads((config_toml_path(tmp_path)).read_text()) assert raw["hub"]["url"] == "https://localhost:1337" 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", "../traversal"]) 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\x00malicious"]) 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 = agent_md_path(tmp_path) 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 = config_toml_path(tmp_path) if config_path.exists(): raw = tomllib.loads(config_path.read_text()) assert "injected" not in raw # --------------------------------------------------------------------------- # Integration — smart sync (skip in-sync files) # --------------------------------------------------------------------------- class TestSmartSync: def test_second_sync_skips_in_sync_files(self, tmp_path: pathlib.Path) -> None: """Repeated sync without changes exits 0 and reports skipped.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 0 assert "in sync" in result.output def test_second_sync_json_skipped_true(self, tmp_path: pathlib.Path) -> None: """sync --json shows skipped=True for already-in-sync files.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "sync", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) for entry in data["adapters"]: assert entry["skipped"] is True assert entry["written"] is False def test_out_of_sync_file_is_updated_without_force(self, tmp_path: pathlib.Path) -> None: """An adapter that is out of sync is updated even without --force.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) # Corrupt AGENTS.md content (tmp_path / "AGENTS.md").write_text("old content") result = _invoke(tmp_path, ["agent-config", "sync"]) assert result.exit_code == 0 assert "old content" not in (tmp_path / "AGENTS.md").read_text() assert "Muse" in (tmp_path / "AGENTS.md").read_text() def test_force_rewrites_even_in_sync_files(self, tmp_path: pathlib.Path) -> None: """--force writes all files even when they are already in sync.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "sync", "--force", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) for entry in data["adapters"]: assert entry["written"] is True assert entry["skipped"] is False def test_sync_idempotent_across_multiple_runs(self, tmp_path: pathlib.Path) -> None: """Running sync N times produces identical output each time.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) content_after_first = (tmp_path / "AGENTS.md").read_text() for _ in range(5): r = _invoke(tmp_path, ["agent-config", "sync"]) assert r.exit_code == 0 assert (tmp_path / "AGENTS.md").read_text() == content_after_first # --------------------------------------------------------------------------- # Integration — inspect # --------------------------------------------------------------------------- class TestInspect: def test_inspect_json_schema_standalone(self, tmp_path: pathlib.Path) -> None: """inspect --json returns all required fields for a standalone repo.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["context"] == "standalone" assert data["workspace_root"] is None assert data["repo_name"] == tmp_path.name assert data["agent_md_exists"] is True assert data["merged_content"] is not None assert "adapters" in data assert isinstance(data["ready"], bool) def test_inspect_ready_true_when_in_sync(self, tmp_path: pathlib.Path) -> None: """ready is True when agent.md exists and adapters are in sync.""" _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) assert data["ready"] is True def test_inspect_ready_false_without_adapters(self, tmp_path: pathlib.Path) -> None: """ready is False when agent.md exists but no adapters have been synced.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) assert data["ready"] is False def test_inspect_ready_false_without_agent_md(self, tmp_path: pathlib.Path) -> None: """ready is False when agent.md does not exist.""" _init_repo(tmp_path) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) assert data["ready"] is False assert data["agent_md_exists"] is False assert data["merged_content"] is None def test_inspect_merged_content_contains_rules(self, tmp_path: pathlib.Path) -> None: """merged_content includes the actual rules from agent.md.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) assert "Muse" in data["merged_content"] assert "git" in data["merged_content"].lower() def test_inspect_adapter_entries_schema(self, tmp_path: pathlib.Path) -> None: """Each adapter entry in inspect output has the expected fields.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) for entry in data["adapters"]: assert "name" in entry assert "filename" in entry assert "exists" in entry assert "in_sync" in entry def test_inspect_workspace_member_context(self, tmp_path: pathlib.Path) -> None: """inspect reports workspace_member context and non-null workspace_root.""" _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", "inspect", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["context"] == "workspace_member" assert data["workspace_root"] is not None assert data["repo_name"] == "core" def test_inspect_workspace_merged_content_includes_both_levels( self, tmp_path: pathlib.Path ) -> None: """merged_content in a workspace member includes both WS and repo rules.""" _init_workspace(tmp_path, [("core", "core")]) _invoke(tmp_path, ["agent-config", "init"]) # Add a unique marker to the workspace-level agent.md ws_agent = agent_md_path(tmp_path) ws_agent.write_text(f"{ws_agent.read_text()}\n# WS_MARKER\n") repo = tmp_path / "core" repo.mkdir() _init_repo(repo) _invoke(repo, ["agent-config", "init"]) # Add a unique marker to the repo-level agent.md repo_agent = agent_md_path(repo) repo_agent.write_text(f"{repo_agent.read_text()}\n# REPO_MARKER\n") result = _invoke(repo, ["agent-config", "inspect", "--json"]) data = json.loads(result.output) assert "WS_MARKER" in data["merged_content"] assert "REPO_MARKER" in data["merged_content"] def test_inspect_text_output_exits_0(self, tmp_path: pathlib.Path) -> None: """inspect without --json exits 0 and prints context info.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "inspect"]) assert result.exit_code == 0 assert "Context" in result.output or "standalone" in result.output # --------------------------------------------------------------------------- # Integration — status extra fields # --------------------------------------------------------------------------- class TestStatusExtraFields: def test_status_json_includes_agent_md_exists(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(result.output) assert "agent_md_exists" in data assert data["agent_md_exists"] is True def test_status_json_includes_ready(self, tmp_path: pathlib.Path) -> None: _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(result.output) assert "ready" in data assert data["ready"] is True def test_status_json_ready_false_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", "--json"]) data = json.loads(result.output) assert data["ready"] is False def test_status_json_summary_counts(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _ADAPTERS _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) result = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(result.output) assert "in_sync_count" in data assert "missing_count" in data assert "out_of_sync_count" in data # Before sync, all adapters are missing assert data["missing_count"] == len(_ADAPTERS) assert data["in_sync_count"] == 0 def test_status_json_counts_after_sync(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.agent_config import _ADAPTERS _init_with_all_adapters(tmp_path) _invoke(tmp_path, ["agent-config", "sync"]) result = _invoke(tmp_path, ["agent-config", "status", "--json"]) data = json.loads(result.output) assert data["in_sync_count"] == len(_ADAPTERS) assert data["missing_count"] == 0 assert data["out_of_sync_count"] == 0 # --------------------------------------------------------------------------- # Integration — template content # --------------------------------------------------------------------------- class TestTemplateContent: def test_standalone_template_includes_testing_rules( self, tmp_path: pathlib.Path ) -> None: """Standalone template includes the no-full-test-suite rule.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) content = (agent_md_path(tmp_path)).read_text() assert "full test suite" in content.lower() or "full" in content.lower() assert "muse code test" in content def test_standalone_template_includes_muse_code_test( self, tmp_path: pathlib.Path ) -> None: """Standalone template lists muse code test in the code intelligence table.""" _init_repo(tmp_path) _invoke(tmp_path, ["agent-config", "init"]) content = (agent_md_path(tmp_path)).read_text() assert "muse code test" in content class TestRegisterFlags: """Argparse registration tests for ``muse agent-config`` subcommands.""" def _parse(self, *args: str) -> argparse.Namespace: from muse.cli.commands.agent_config import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["agent-config", *args]) # init def test_init_default_json_out_is_false(self) -> None: ns = self._parse("init") assert ns.json_out is False def test_init_json_flag_sets_json_out(self) -> None: ns = self._parse("init", "--json") assert ns.json_out is True def test_init_j_shorthand_sets_json_out(self) -> None: ns = self._parse("init", "-j") assert ns.json_out is True def test_init_force_default(self) -> None: ns = self._parse("init") assert ns.force is False def test_init_force_flag(self) -> None: ns = self._parse("init", "--force") assert ns.force is True def test_init_force_shorthand(self) -> None: ns = self._parse("init", "-f") assert ns.force is True # sync def test_sync_default_json_out_is_false(self) -> None: ns = self._parse("sync") assert ns.json_out is False def test_sync_json_flag_sets_json_out(self) -> None: ns = self._parse("sync", "--json") assert ns.json_out is True def test_sync_j_shorthand_sets_json_out(self) -> None: ns = self._parse("sync", "-j") assert ns.json_out is True def test_sync_dry_run_default(self) -> None: ns = self._parse("sync") assert ns.dry_run is False def test_sync_dry_run_flag(self) -> None: ns = self._parse("sync", "--dry-run") assert ns.dry_run is True def test_sync_dry_run_shorthand(self) -> None: ns = self._parse("sync", "-n") assert ns.dry_run is True def test_sync_force_default(self) -> None: ns = self._parse("sync") assert ns.force is False def test_sync_force_flag(self) -> None: ns = self._parse("sync", "--force") assert ns.force is True def test_sync_force_shorthand(self) -> None: ns = self._parse("sync", "-f") assert ns.force is True # read def test_read_default_json_out_is_false(self) -> None: ns = self._parse("read") assert ns.json_out is False def test_read_json_flag_sets_json_out(self) -> None: ns = self._parse("read", "--json") assert ns.json_out is True def test_read_j_shorthand_sets_json_out(self) -> None: ns = self._parse("read", "-j") assert ns.json_out is True # status def test_status_default_json_out_is_false(self) -> None: ns = self._parse("status") assert ns.json_out is False def test_status_json_flag_sets_json_out(self) -> None: ns = self._parse("status", "--json") assert ns.json_out is True # inspect def test_inspect_default_json_out_is_false(self) -> None: ns = self._parse("inspect") assert ns.json_out is False def test_inspect_json_flag_sets_json_out(self) -> None: ns = self._parse("inspect", "--json") assert ns.json_out is True # set def test_set_default_json_out_is_false(self) -> None: ns = self._parse("set", "--adapters", "claude") assert ns.json_out is False def test_set_json_flag_sets_json_out(self) -> None: ns = self._parse("set", "--adapters", "claude", "--json") assert ns.json_out is True