gabriel / muse public
test_cmd_config.py python
349 lines 16.2 KB
Raw
86000da9 fix: replace typer CliRunner with argparse-compatible test helper Gabriel Cardona <gabriel@tellurstori.com> 1d 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 import uuid
16
17 import pytest
18 from tests.cli_test_helper import CliRunner
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 = str(uuid.uuid4())
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) -> dict[str, str]:
49 return {"MUSE_REPO_ROOT": str(root)}
50
51
52 # ---------------------------------------------------------------------------
53 # Unit tests — config helpers
54 # ---------------------------------------------------------------------------
55
56
57 class TestConfigValueHelpers:
58 def test_set_and_get_user_name(self, tmp_path: pathlib.Path) -> None:
59 root, _ = _init_repo(tmp_path)
60 from muse.cli.config import get_config_value, set_config_value
61 set_config_value("user.name", "Alice", root)
62 assert get_config_value("user.name", root) == "Alice"
63
64 def test_set_and_get_user_email(self, tmp_path: pathlib.Path) -> None:
65 root, _ = _init_repo(tmp_path)
66 from muse.cli.config import get_config_value, set_config_value
67 set_config_value("user.email", "alice@example.com", root)
68 assert get_config_value("user.email", root) == "alice@example.com"
69
70 def test_set_and_get_user_type(self, tmp_path: pathlib.Path) -> None:
71 root, _ = _init_repo(tmp_path)
72 from muse.cli.config import get_config_value, set_config_value
73 set_config_value("user.type", "agent", root)
74 assert get_config_value("user.type", root) == "agent"
75
76 def test_set_and_get_domain_key(self, tmp_path: pathlib.Path) -> None:
77 root, _ = _init_repo(tmp_path)
78 from muse.cli.config import get_config_value, set_config_value
79 set_config_value("domain.ticks_per_beat", "480", root)
80 assert get_config_value("domain.ticks_per_beat", root) == "480"
81
82 def test_get_missing_key_returns_none(self, tmp_path: pathlib.Path) -> None:
83 root, _ = _init_repo(tmp_path)
84 from muse.cli.config import get_config_value
85 assert get_config_value("user.name", root) is None
86
87 def test_get_unknown_namespace_returns_none(self, tmp_path: pathlib.Path) -> None:
88 root, _ = _init_repo(tmp_path)
89 from muse.cli.config import get_config_value
90 assert get_config_value("unknown.key", root) is None
91
92 def test_set_blocked_auth_raises(self, tmp_path: pathlib.Path) -> None:
93 root, _ = _init_repo(tmp_path)
94 from muse.cli.config import set_config_value
95 with pytest.raises(ValueError, match="muse auth login"):
96 set_config_value("auth.token", "secret", root)
97
98 def test_set_blocked_remotes_raises(self, tmp_path: pathlib.Path) -> None:
99 root, _ = _init_repo(tmp_path)
100 from muse.cli.config import set_config_value
101 with pytest.raises(ValueError, match="muse remote"):
102 set_config_value("remotes.origin", "https://x.com", root)
103
104 def test_set_unknown_namespace_raises(self, tmp_path: pathlib.Path) -> None:
105 root, _ = _init_repo(tmp_path)
106 from muse.cli.config import set_config_value
107 with pytest.raises(ValueError):
108 set_config_value("invalid.key", "value", root)
109
110 def test_set_malformed_key_raises(self, tmp_path: pathlib.Path) -> None:
111 root, _ = _init_repo(tmp_path)
112 from muse.cli.config import set_config_value
113 with pytest.raises(ValueError):
114 set_config_value("no-dot-key", "value", root)
115
116 def test_config_as_dict_includes_user(self, tmp_path: pathlib.Path) -> None:
117 root, _ = _init_repo(tmp_path)
118 from muse.cli.config import config_as_dict, set_config_value
119 set_config_value("user.name", "Bob", root)
120 d = config_as_dict(root)
121 assert d.get("user", {}).get("name") == "Bob"
122
123 def test_config_as_dict_empty_repo(self, tmp_path: pathlib.Path) -> None:
124 root, _ = _init_repo(tmp_path)
125 from muse.cli.config import config_as_dict
126 d = config_as_dict(root)
127 assert isinstance(d, dict)
128
129 def test_set_hub_url_requires_https(self, tmp_path: pathlib.Path) -> None:
130 root, _ = _init_repo(tmp_path)
131 from muse.cli.config import set_config_value
132 with pytest.raises(ValueError, match="HTTPS"):
133 set_config_value("hub.url", "http://insecure.example.com", root)
134
135
136 # ---------------------------------------------------------------------------
137 # Integration tests — CLI commands
138 # ---------------------------------------------------------------------------
139
140
141 class TestConfigCLI:
142 def test_show_empty_config(self, tmp_path: pathlib.Path) -> None:
143 root, _ = _init_repo(tmp_path)
144 result = runner.invoke(cli, ["config", "show"], env=_env(root), catch_exceptions=False)
145 assert result.exit_code == 0
146
147 def test_show_json_empty(self, tmp_path: pathlib.Path) -> None:
148 root, _ = _init_repo(tmp_path)
149 result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root), catch_exceptions=False)
150 assert result.exit_code == 0
151 data = json.loads(result.output)
152 assert isinstance(data, dict)
153
154 def test_show_format_json(self, tmp_path: pathlib.Path) -> None:
155 root, _ = _init_repo(tmp_path)
156 result = runner.invoke(cli, ["config", "show", "--format", "json"], env=_env(root), catch_exceptions=False)
157 assert result.exit_code == 0
158 data = json.loads(result.output)
159 assert isinstance(data, dict)
160
161 def test_set_user_name(self, tmp_path: pathlib.Path) -> None:
162 root, _ = _init_repo(tmp_path)
163 result = runner.invoke(cli, ["config", "set", "user.name", "Alice"], env=_env(root), catch_exceptions=False)
164 assert result.exit_code == 0
165 assert "user.name" in result.output
166
167 def test_set_then_get_user_name(self, tmp_path: pathlib.Path) -> None:
168 root, _ = _init_repo(tmp_path)
169 runner.invoke(cli, ["config", "set", "user.name", "Carol"], env=_env(root), catch_exceptions=False)
170 result = runner.invoke(cli, ["config", "get", "user.name"], env=_env(root), catch_exceptions=False)
171 assert result.exit_code == 0
172 assert "Carol" in result.output
173
174 def test_get_unset_key_fails(self, tmp_path: pathlib.Path) -> None:
175 root, _ = _init_repo(tmp_path)
176 result = runner.invoke(cli, ["config", "get", "user.name"], env=_env(root))
177 assert result.exit_code != 0
178
179 def test_set_domain_key(self, tmp_path: pathlib.Path) -> None:
180 root, _ = _init_repo(tmp_path)
181 result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"],
182 env=_env(root), catch_exceptions=False)
183 assert result.exit_code == 0
184
185 def test_get_domain_key_after_set(self, tmp_path: pathlib.Path) -> None:
186 root, _ = _init_repo(tmp_path)
187 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"], env=_env(root))
188 result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"], env=_env(root), catch_exceptions=False)
189 assert result.exit_code == 0
190 assert "960" in result.output
191
192 def test_set_blocked_auth_fails(self, tmp_path: pathlib.Path) -> None:
193 root, _ = _init_repo(tmp_path)
194 result = runner.invoke(cli, ["config", "set", "auth.token", "secret"], env=_env(root))
195 assert result.exit_code != 0
196
197 def test_set_blocked_remotes_fails(self, tmp_path: pathlib.Path) -> None:
198 root, _ = _init_repo(tmp_path)
199 result = runner.invoke(cli, ["config", "set", "remotes.origin", "https://x.com"], env=_env(root))
200 assert result.exit_code != 0
201
202 def test_set_http_hub_url_fails(self, tmp_path: pathlib.Path) -> None:
203 root, _ = _init_repo(tmp_path)
204 result = runner.invoke(cli, ["config", "set", "hub.url", "http://insecure.example.com"], env=_env(root))
205 assert result.exit_code != 0
206
207 def test_set_https_hub_url_succeeds(self, tmp_path: pathlib.Path) -> None:
208 root, _ = _init_repo(tmp_path)
209 result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"],
210 env=_env(root), catch_exceptions=False)
211 assert result.exit_code == 0
212
213 def test_show_after_set_includes_value(self, tmp_path: pathlib.Path) -> None:
214 root, _ = _init_repo(tmp_path)
215 runner.invoke(cli, ["config", "set", "user.name", "Dave"], env=_env(root))
216 result = runner.invoke(cli, ["config", "show"], env=_env(root), catch_exceptions=False)
217 assert result.exit_code == 0
218 assert "Dave" in result.output
219
220 def test_show_json_after_set(self, tmp_path: pathlib.Path) -> None:
221 root, _ = _init_repo(tmp_path)
222 runner.invoke(cli, ["config", "set", "user.name", "Eve"], env=_env(root))
223 runner.invoke(cli, ["config", "set", "user.type", "agent"], env=_env(root))
224 result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root), catch_exceptions=False)
225 assert result.exit_code == 0
226 data = json.loads(result.output)
227 assert data.get("user", {}).get("name") == "Eve"
228 assert data.get("user", {}).get("type") == "agent"
229
230 def test_multiple_sets_accumulate(self, tmp_path: pathlib.Path) -> None:
231 root, _ = _init_repo(tmp_path)
232 runner.invoke(cli, ["config", "set", "user.name", "Frank"], env=_env(root))
233 runner.invoke(cli, ["config", "set", "user.email", "frank@example.com"], env=_env(root))
234 runner.invoke(cli, ["config", "set", "domain.key", "val"], env=_env(root))
235 result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root), catch_exceptions=False)
236 data = json.loads(result.output)
237 assert data["user"]["name"] == "Frank"
238 assert data["user"]["email"] == "frank@example.com"
239 assert data["domain"]["key"] == "val"
240
241 def test_set_overwrites_previous_value(self, tmp_path: pathlib.Path) -> None:
242 root, _ = _init_repo(tmp_path)
243 runner.invoke(cli, ["config", "set", "user.name", "Old"], env=_env(root))
244 runner.invoke(cli, ["config", "set", "user.name", "New"], env=_env(root))
245 result = runner.invoke(cli, ["config", "get", "user.name"], env=_env(root), catch_exceptions=False)
246 assert result.exit_code == 0
247 assert "New" in result.output
248
249 def test_show_format_unknown_fails(self, tmp_path: pathlib.Path) -> None:
250 root, _ = _init_repo(tmp_path)
251 result = runner.invoke(cli, ["config", "show", "--format", "xml"], env=_env(root))
252 assert result.exit_code != 0
253
254
255 # ---------------------------------------------------------------------------
256 # E2E tests
257 # ---------------------------------------------------------------------------
258
259
260 class TestConfigE2E:
261 def test_full_agent_config_workflow(self, tmp_path: pathlib.Path) -> None:
262 """Agent sets identity, then reads it back as JSON."""
263 root, _ = _init_repo(tmp_path)
264 runner.invoke(cli, ["config", "set", "user.name", "muse-agent-001"], env=_env(root))
265 runner.invoke(cli, ["config", "set", "user.type", "agent"], env=_env(root))
266 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"], env=_env(root))
267
268 result = runner.invoke(cli, ["config", "show", "--format", "json"], env=_env(root), catch_exceptions=False)
269 assert result.exit_code == 0
270 data = json.loads(result.output)
271 assert data["user"]["name"] == "muse-agent-001"
272 assert data["user"]["type"] == "agent"
273 assert data["domain"]["ticks_per_beat"] == "960"
274
275 def test_config_persists_across_invocations(self, tmp_path: pathlib.Path) -> None:
276 """Config written in one invocation is readable in a subsequent one."""
277 root, _ = _init_repo(tmp_path)
278 runner.invoke(cli, ["config", "set", "user.name", "Persistent"], env=_env(root))
279 result = runner.invoke(cli, ["config", "get", "user.name"], env=_env(root), catch_exceptions=False)
280 assert "Persistent" in result.output
281
282
283 # ---------------------------------------------------------------------------
284 # Security tests
285 # ---------------------------------------------------------------------------
286
287
288 class TestConfigSecurity:
289 def test_toml_injection_in_name_is_stored_safely(self, tmp_path: pathlib.Path) -> None:
290 """TOML injection chars in a name value do not break the config file."""
291 root, _ = _init_repo(tmp_path)
292 injection = 'Alice"\n[injected]\nkey = "value'
293 result = runner.invoke(cli, ["config", "set", "user.name", injection], env=_env(root))
294 # Should either fail safely or store the value escaped
295 if result.exit_code == 0:
296 get_result = runner.invoke(cli, ["config", "get", "user.name"], env=_env(root))
297 # If stored, round-trip must be stable — no config file corruption
298 show_result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root))
299 assert show_result.exit_code == 0
300 data = json.loads(show_result.output)
301 assert isinstance(data, dict)
302
303 def test_no_credentials_in_json_output(self, tmp_path: pathlib.Path) -> None:
304 """config show --json never leaks credentials even if they somehow end up in config.toml."""
305 root, _ = _init_repo(tmp_path)
306 config_path = root / ".muse" / "config.toml"
307 # Manually inject a fake token into config.toml
308 config_path.write_text('[auth]\ntoken = "super-secret"\n', encoding="utf-8")
309 result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root), catch_exceptions=False)
310 assert result.exit_code == 0
311 assert "super-secret" not in result.output
312
313 def test_set_user_type_rejects_unknown_values_gracefully(self, tmp_path: pathlib.Path) -> None:
314 """user.type accepts free-form values — but they are stored, not validated."""
315 root, _ = _init_repo(tmp_path)
316 result = runner.invoke(cli, ["config", "set", "user.type", "robot"], env=_env(root), catch_exceptions=False)
317 # Current behaviour: stored as-is. This tests it doesn't crash.
318 assert result.exit_code == 0
319
320
321 # ---------------------------------------------------------------------------
322 # Stress tests
323 # ---------------------------------------------------------------------------
324
325
326 class TestConfigStress:
327 def test_many_domain_keys(self, tmp_path: pathlib.Path) -> None:
328 """Setting 50 domain keys all survive a JSON round-trip."""
329 root, _ = _init_repo(tmp_path)
330 keys = {f"domain.key_{i}": str(i) for i in range(50)}
331 for k, v in keys.items():
332 r = runner.invoke(cli, ["config", "set", k, v], env=_env(root))
333 assert r.exit_code == 0
334
335 result = runner.invoke(cli, ["config", "show", "--json"], env=_env(root), catch_exceptions=False)
336 assert result.exit_code == 0
337 data = json.loads(result.output)
338 domain = data.get("domain", {})
339 for i in range(50):
340 assert domain.get(f"key_{i}") == str(i)
341
342 def test_overwrite_domain_key_many_times(self, tmp_path: pathlib.Path) -> None:
343 """Repeated writes to the same key keep only the latest value."""
344 root, _ = _init_repo(tmp_path)
345 for i in range(20):
346 runner.invoke(cli, ["config", "set", "domain.counter", str(i)], env=_env(root))
347 result = runner.invoke(cli, ["config", "get", "domain.counter"], env=_env(root), catch_exceptions=False)
348 assert result.exit_code == 0
349 assert "19" in result.output