gabriel / muse public
test_cmd_config.py python
349 lines 16.2 KB
Raw
faec8c4d feat(hardening): add config/bisect CLI tests and fix bisect convergence bug Gabriel Cardona <gabriel@tellurstori.com> 2d 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 typer.testing import CliRunner
19
20 from muse.cli.app import cli
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