gabriel / muse public
test_cmd_config.py python
400 lines 18.1 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Comprehensive tests for ``muse config`` — show / get / set.
2
3 Coverage:
4 - Unit: get_config_value, set_config_value, config_as_dict
5 - Integration: CLI round-trips for show, get, set
6 - E2E: full set→get→show workflow
7 - Security: blocked namespaces, TOML injection, malformed keys
8 - Format: --json / --format json output
9 """
10
11 from __future__ import annotations
12
13 import json
14 import pathlib
15
16 import pytest
17 from tests.cli_test_helper import CliRunner
18 from muse.core._types import fake_id
19
20 cli = None # argparse migration — CliRunner ignores this arg
21
22 runner = CliRunner()
23
24
25 # ---------------------------------------------------------------------------
26 # Helpers
27 # ---------------------------------------------------------------------------
28
29
30 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
31 """Initialise a minimal .muse repo and return (root, repo_id)."""
32 repo_id = fake_id("repo")
33 muse = tmp_path / ".muse"
34 muse.mkdir()
35 (muse / "repo.json").write_text(
36 json.dumps({"repo_id": repo_id, "domain": "midi",
37 "default_branch": "main",
38 "created_at": "2026-01-01T00:00:00+00:00"})
39 )
40 (muse / "HEAD").write_text("ref: refs/heads/main")
41 (muse / "refs" / "heads").mkdir(parents=True)
42 (muse / "snapshots").mkdir()
43 (muse / "commits").mkdir()
44 (muse / "objects").mkdir()
45 return tmp_path, repo_id
46
47
48 def _env(root: pathlib.Path) -> Manifest:
49 return {"MUSE_REPO_ROOT": str(root)}
50
51
52 # ---------------------------------------------------------------------------
53 # Parser flag tests
54 # ---------------------------------------------------------------------------
55
56
57 class TestRegisterFlags:
58 def _parse(self, *args: str) -> "argparse.Namespace":
59 import argparse
60 from muse.cli.commands.config_cmd import register
61 p = argparse.ArgumentParser()
62 sub = p.add_subparsers()
63 register(sub)
64 return p.parse_args(["config", *args])
65
66 def test_get_default_json_out_is_false(self) -> None:
67 ns = self._parse("get", "user.handle")
68 assert ns.json_out is False
69
70 def test_get_json_flag_sets_json_out(self) -> None:
71 ns = self._parse("get", "user.handle", "--json")
72 assert ns.json_out is True
73
74 def test_get_j_shorthand_sets_json_out(self) -> None:
75 ns = self._parse("get", "user.handle", "-j")
76 assert ns.json_out is True
77
78 def test_set_default_json_out_is_false(self) -> None:
79 ns = self._parse("set", "user.handle", "Alice")
80 assert ns.json_out is False
81
82 def test_set_json_flag_sets_json_out(self) -> None:
83 ns = self._parse("set", "user.handle", "Alice", "--json")
84 assert ns.json_out is True
85
86 def test_read_default_json_out_is_false(self) -> None:
87 ns = self._parse("read")
88 assert ns.json_out is False
89
90 def test_read_json_flag_sets_json_out(self) -> None:
91 ns = self._parse("read", "--json")
92 assert ns.json_out is True
93
94 def test_read_j_shorthand_sets_json_out(self) -> None:
95 ns = self._parse("read", "-j")
96 assert ns.json_out is True
97
98
99 # ---------------------------------------------------------------------------
100 # Unit tests — config helpers
101 # ---------------------------------------------------------------------------
102
103
104 class TestConfigValueHelpers:
105 def test_set_and_get_user_name(self, tmp_path: pathlib.Path) -> None:
106 root, _ = _init_repo(tmp_path)
107 from muse.cli.config import get_config_value, set_config_value
108 set_config_value("user.handle", "Alice", root)
109 assert get_config_value("user.handle", root) == "Alice"
110
111 def test_set_and_get_user_email(self, tmp_path: pathlib.Path) -> None:
112 root, _ = _init_repo(tmp_path)
113 from muse.cli.config import get_config_value, set_config_value
114 set_config_value("user.email", "[email protected]", root)
115 assert get_config_value("user.email", root) == "[email protected]"
116
117 def test_set_and_get_user_type(self, tmp_path: pathlib.Path) -> None:
118 root, _ = _init_repo(tmp_path)
119 from muse.cli.config import get_config_value, set_config_value
120 set_config_value("user.type", "agent", root)
121 assert get_config_value("user.type", root) == "agent"
122
123 def test_set_and_get_domain_key(self, tmp_path: pathlib.Path) -> None:
124 root, _ = _init_repo(tmp_path)
125 from muse.cli.config import get_config_value, set_config_value
126 set_config_value("domain.ticks_per_beat", "480", root)
127 assert get_config_value("domain.ticks_per_beat", root) == "480"
128
129 def test_get_missing_key_returns_none(self, tmp_path: pathlib.Path) -> None:
130 root, _ = _init_repo(tmp_path)
131 from muse.cli.config import get_config_value
132 assert get_config_value("user.handle", root) is None
133
134 def test_get_unknown_namespace_returns_none(self, tmp_path: pathlib.Path) -> None:
135 root, _ = _init_repo(tmp_path)
136 from muse.cli.config import get_config_value
137 assert get_config_value("unknown.key", root) is None
138
139 def test_set_blocked_auth_raises(self, tmp_path: pathlib.Path) -> None:
140 root, _ = _init_repo(tmp_path)
141 from muse.cli.config import set_config_value
142 with pytest.raises(ValueError, match="muse auth keygen"):
143 set_config_value("auth.anything", "secret", root)
144
145 def test_set_blocked_remotes_raises(self, tmp_path: pathlib.Path) -> None:
146 root, _ = _init_repo(tmp_path)
147 from muse.cli.config import set_config_value
148 with pytest.raises(ValueError, match="muse remote"):
149 set_config_value("remotes.origin", "https://x.com", root)
150
151 def test_set_unknown_namespace_raises(self, tmp_path: pathlib.Path) -> None:
152 root, _ = _init_repo(tmp_path)
153 from muse.cli.config import set_config_value
154 with pytest.raises(ValueError):
155 set_config_value("invalid.key", "value", root)
156
157 def test_set_malformed_key_raises(self, tmp_path: pathlib.Path) -> None:
158 root, _ = _init_repo(tmp_path)
159 from muse.cli.config import set_config_value
160 with pytest.raises(ValueError):
161 set_config_value("no-dot-key", "value", root)
162
163 def test_config_as_dict_includes_user(self, tmp_path: pathlib.Path) -> None:
164 root, _ = _init_repo(tmp_path)
165 from muse.cli.config import config_as_dict, set_config_value
166 set_config_value("user.handle", "Bob", root)
167 d = config_as_dict(root)
168 assert d.get("user", {}).get("handle") == "Bob"
169
170 def test_config_as_dict_empty_repo(self, tmp_path: pathlib.Path) -> None:
171 root, _ = _init_repo(tmp_path)
172 from muse.cli.config import config_as_dict
173 d = config_as_dict(root)
174 assert isinstance(d, dict)
175
176 def test_set_hub_url_requires_https(self, tmp_path: pathlib.Path) -> None:
177 root, _ = _init_repo(tmp_path)
178 from muse.cli.config import set_config_value
179 with pytest.raises(ValueError, match="HTTPS"):
180 set_config_value("hub.url", "http://insecure.example.com", root)
181
182
183 # ---------------------------------------------------------------------------
184 # Integration tests — CLI commands
185 # ---------------------------------------------------------------------------
186
187
188 class TestConfigCLI:
189 def test_read_empty_config(self, tmp_path: pathlib.Path) -> None:
190 root, _ = _init_repo(tmp_path)
191 result = runner.invoke(cli, ["config", "read"], env=_env(root), catch_exceptions=False)
192 assert result.exit_code == 0
193
194 def test_read_json_empty(self, tmp_path: pathlib.Path) -> None:
195 root, _ = _init_repo(tmp_path)
196 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
197 assert result.exit_code == 0
198 data = json.loads(result.output)
199 assert isinstance(data, dict)
200
201 def test_read_format_json(self, tmp_path: pathlib.Path) -> None:
202 root, _ = _init_repo(tmp_path)
203 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
204 assert result.exit_code == 0
205 data = json.loads(result.output)
206 assert isinstance(data, dict)
207
208 def test_set_user_name(self, tmp_path: pathlib.Path) -> None:
209 root, _ = _init_repo(tmp_path)
210 result = runner.invoke(cli, ["config", "set", "user.handle", "Alice"], env=_env(root), catch_exceptions=False)
211 assert result.exit_code == 0
212 assert "user.handle" in result.output
213
214 def test_set_then_get_user_name(self, tmp_path: pathlib.Path) -> None:
215 root, _ = _init_repo(tmp_path)
216 runner.invoke(cli, ["config", "set", "user.handle", "Carol"], env=_env(root), catch_exceptions=False)
217 result = runner.invoke(cli, ["config", "get", "user.handle"], env=_env(root), catch_exceptions=False)
218 assert result.exit_code == 0
219 assert "Carol" in result.output
220
221 def test_get_unset_key_fails(self, tmp_path: pathlib.Path) -> None:
222 root, _ = _init_repo(tmp_path)
223 result = runner.invoke(cli, ["config", "get", "user.handle"], env=_env(root))
224 assert result.exit_code != 0
225
226 def test_set_domain_key(self, tmp_path: pathlib.Path) -> None:
227 root, _ = _init_repo(tmp_path)
228 result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"],
229 env=_env(root), catch_exceptions=False)
230 assert result.exit_code == 0
231
232 def test_get_domain_key_after_set(self, tmp_path: pathlib.Path) -> None:
233 root, _ = _init_repo(tmp_path)
234 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"], env=_env(root))
235 result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"], env=_env(root), catch_exceptions=False)
236 assert result.exit_code == 0
237 assert "960" in result.output
238
239 def test_set_blocked_auth_fails(self, tmp_path: pathlib.Path) -> None:
240 root, _ = _init_repo(tmp_path)
241 result = runner.invoke(cli, ["config", "set", "auth.anything", "secret"], env=_env(root))
242 assert result.exit_code != 0
243
244 def test_set_blocked_remotes_fails(self, tmp_path: pathlib.Path) -> None:
245 root, _ = _init_repo(tmp_path)
246 result = runner.invoke(cli, ["config", "set", "remotes.origin", "https://x.com"], env=_env(root))
247 assert result.exit_code != 0
248
249 def test_set_http_hub_url_fails(self, tmp_path: pathlib.Path) -> None:
250 root, _ = _init_repo(tmp_path)
251 result = runner.invoke(cli, ["config", "set", "hub.url", "http://insecure.example.com"], env=_env(root))
252 assert result.exit_code != 0
253
254 def test_set_https_hub_url_succeeds(self, tmp_path: pathlib.Path) -> None:
255 root, _ = _init_repo(tmp_path)
256 result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"],
257 env=_env(root), catch_exceptions=False)
258 assert result.exit_code == 0
259
260 def test_read_after_set_includes_value(self, tmp_path: pathlib.Path) -> None:
261 root, _ = _init_repo(tmp_path)
262 runner.invoke(cli, ["config", "set", "user.handle", "Dave"], env=_env(root))
263 result = runner.invoke(cli, ["config", "read"], env=_env(root), catch_exceptions=False)
264 assert result.exit_code == 0
265 assert "Dave" in result.output
266
267 def test_read_json_after_set(self, tmp_path: pathlib.Path) -> None:
268 root, _ = _init_repo(tmp_path)
269 runner.invoke(cli, ["config", "set", "user.handle", "Eve"], env=_env(root))
270 runner.invoke(cli, ["config", "set", "user.type", "agent"], env=_env(root))
271 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
272 assert result.exit_code == 0
273 envelope = json.loads(result.output)
274 data = envelope.get("config", envelope)
275 assert data.get("user", {}).get("handle") == "Eve"
276 assert data.get("user", {}).get("type") == "agent"
277
278 def test_multiple_sets_accumulate(self, tmp_path: pathlib.Path) -> None:
279 root, _ = _init_repo(tmp_path)
280 runner.invoke(cli, ["config", "set", "user.handle", "Frank"], env=_env(root))
281 runner.invoke(cli, ["config", "set", "user.email", "[email protected]"], env=_env(root))
282 runner.invoke(cli, ["config", "set", "domain.key", "val"], env=_env(root))
283 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
284 envelope = json.loads(result.output)
285 data = envelope.get("config", envelope)
286 assert data["user"]["handle"] == "Frank"
287 assert data["user"]["email"] == "[email protected]"
288 assert data["domain"]["key"] == "val"
289
290 def test_set_overwrites_previous_value(self, tmp_path: pathlib.Path) -> None:
291 root, _ = _init_repo(tmp_path)
292 runner.invoke(cli, ["config", "set", "user.handle", "Old"], env=_env(root))
293 runner.invoke(cli, ["config", "set", "user.handle", "New"], env=_env(root))
294 result = runner.invoke(cli, ["config", "get", "user.handle"], env=_env(root), catch_exceptions=False)
295 assert result.exit_code == 0
296 assert "New" in result.output
297
298 def test_read_format_unknown_fails(self, tmp_path: pathlib.Path) -> None:
299 root, _ = _init_repo(tmp_path)
300 result = runner.invoke(cli, ["config", "read", "--format", "xml"], env=_env(root))
301 assert result.exit_code != 0
302
303
304 # ---------------------------------------------------------------------------
305 # E2E tests
306 # ---------------------------------------------------------------------------
307
308
309 class TestConfigE2E:
310 def test_full_agent_config_workflow(self, tmp_path: pathlib.Path) -> None:
311 """Agent sets identity, then reads it back as JSON."""
312 root, _ = _init_repo(tmp_path)
313 runner.invoke(cli, ["config", "set", "user.handle", "muse-agent-001"], env=_env(root))
314 runner.invoke(cli, ["config", "set", "user.type", "agent"], env=_env(root))
315 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"], env=_env(root))
316
317 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
318 assert result.exit_code == 0
319 envelope = json.loads(result.output)
320 data = envelope.get("config", envelope)
321 assert data["user"]["handle"] == "muse-agent-001"
322 assert data["user"]["type"] == "agent"
323 assert data["domain"]["ticks_per_beat"] == "960"
324
325 def test_config_persists_across_invocations(self, tmp_path: pathlib.Path) -> None:
326 """Config written in one invocation is readable in a subsequent one."""
327 root, _ = _init_repo(tmp_path)
328 runner.invoke(cli, ["config", "set", "user.handle", "Persistent"], env=_env(root))
329 result = runner.invoke(cli, ["config", "get", "user.handle"], env=_env(root), catch_exceptions=False)
330 assert "Persistent" in result.output
331
332
333 # ---------------------------------------------------------------------------
334 # Security tests
335 # ---------------------------------------------------------------------------
336
337
338 class TestConfigSecurity:
339 def test_toml_injection_in_name_is_stored_safely(self, tmp_path: pathlib.Path) -> None:
340 """TOML injection chars in a name value do not break the config file."""
341 root, _ = _init_repo(tmp_path)
342 injection = 'Alice"\n[injected]\nkey = "value'
343 result = runner.invoke(cli, ["config", "set", "user.handle", injection], env=_env(root))
344 # Should either fail safely or store the value escaped
345 if result.exit_code == 0:
346 get_result = runner.invoke(cli, ["config", "get", "user.handle"], env=_env(root))
347 # If stored, round-trip must be stable — no config file corruption
348 show_result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root))
349 assert show_result.exit_code == 0
350 data = json.loads(show_result.output)
351 assert isinstance(data, dict)
352
353 def test_no_credentials_in_json_output(self, tmp_path: pathlib.Path) -> None:
354 """config read --json never leaks credentials even if they somehow end up in config.toml."""
355 root, _ = _init_repo(tmp_path)
356 config_path = root / ".muse" / "config.toml"
357 # Manually inject a fake token into config.toml
358 config_path.write_text('[auth]\ntoken = "super-secret"\n', encoding="utf-8")
359 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
360 assert result.exit_code == 0
361 assert "super-secret" not in result.output
362
363 def test_set_user_type_rejects_unknown_values_gracefully(self, tmp_path: pathlib.Path) -> None:
364 """user.type accepts free-form values — but they are stored, not validated."""
365 root, _ = _init_repo(tmp_path)
366 result = runner.invoke(cli, ["config", "set", "user.type", "robot"], env=_env(root), catch_exceptions=False)
367 # Current behaviour: stored as-is. This tests it doesn't crash.
368 assert result.exit_code == 0
369
370
371 # ---------------------------------------------------------------------------
372 # Stress tests
373 # ---------------------------------------------------------------------------
374
375
376 class TestConfigStress:
377 def test_many_domain_keys(self, tmp_path: pathlib.Path) -> None:
378 """Setting 50 domain keys all survive a JSON round-trip."""
379 root, _ = _init_repo(tmp_path)
380 keys = {f"domain.key_{i}": str(i) for i in range(50)}
381 for k, v in keys.items():
382 r = runner.invoke(cli, ["config", "set", k, v], env=_env(root))
383 assert r.exit_code == 0
384
385 result = runner.invoke(cli, ["config", "read", "--json"], env=_env(root), catch_exceptions=False)
386 assert result.exit_code == 0
387 envelope = json.loads(result.output)
388 data = envelope.get("config", envelope)
389 domain = data.get("domain", {})
390 for i in range(50):
391 assert domain.get(f"key_{i}") == str(i)
392
393 def test_overwrite_domain_key_many_times(self, tmp_path: pathlib.Path) -> None:
394 """Repeated writes to the same key keep only the latest value."""
395 root, _ = _init_repo(tmp_path)
396 for i in range(20):
397 runner.invoke(cli, ["config", "set", "domain.counter", str(i)], env=_env(root))
398 result = runner.invoke(cli, ["config", "get", "domain.counter"], env=_env(root), catch_exceptions=False)
399 assert result.exit_code == 0
400 assert "19" in result.output
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 140 days ago