gabriel / muse public
test_cmd_config_hardening.py python
1,808 lines 79.1 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive hardening tests for ``muse config``.
2
3 Coverage
4 --------
5 Unit — muse/cli/config.py internals
6 - _escape: backslash, quote, newline, carriage-return, null-byte escaping
7 - _validate_toml_key: unsafe characters rejected, safe keys pass
8 - _dump_toml: limits section round-trips, domain key injection blocked,
9 remote name injection blocked
10 - set_config_value: limits namespace writes correctly, TOML key injection
11 blocked, unknown limits key rejected, non-integer limits rejected,
12 invalid shard_prefix_length rejected
13 - get_config_value: limits keys read back correctly after write
14 - config_as_dict: limits section included in output
15
16 Integration — CLI commands via CliRunner
17 - run_show: TOML text and JSON outputs, limits displayed in both formats,
18 sanitize_display applied in text mode, --format json alias
19 - run_get: bare value to stdout, --json schema, not-set exits nonzero,
20 key sanitized in stderr message
21 - run_set: success to stderr, --json schema, blocked namespace rejected,
22 TOML injection rejected, limits set and readable, non-integer limits rejected
23 - run_edit: no-repo exits, missing config exits, bad editor exits
24
25 Security
26 - TOML key injection: newline, bracket, equals, quote in domain key blocked
27 - ANSI in key/value sanitized in run_show text mode
28 - ANSI in exception message sanitized in run_set stderr
29 - run_set success message goes to stderr (stdout clean for scripting)
30 - run_get error goes to stderr
31
32 E2E (full round-trip via CLI)
33 - set then get is consistent
34 - set limits then show --json contains limits
35 - set domain then show TOML is valid TOML
36 - limits fall-through to domain is fixed (writes to [limits] not [domain])
37
38 Stress
39 - 8 concurrent set_config_value calls to isolated repos: no corruption
40 """
41
42 from __future__ import annotations
43
44 import json
45 import pathlib
46 import threading
47 import tomllib
48 from typing import TYPE_CHECKING
49 from unittest.mock import MagicMock, patch
50
51 import pytest
52
53 from tests.cli_test_helper import CliRunner, InvokeResult
54
55 if TYPE_CHECKING:
56 pass
57
58 from muse.cli.commands.config_cmd import _GetJson, _SetJson
59 from muse.core.store import JsonValue
60
61 type _ShowJson = dict[str, JsonValue]
62 from muse.core._types import MsgpackDict
63
64 cli = None
65 runner = CliRunner()
66
67 # ── fixtures ──────────────────────────────────────────────────────────────────
68
69
70 @pytest.fixture
71 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
72 """Minimal .muse/ repo with an empty config.toml."""
73 from muse._version import __version__
74
75 muse_dir = tmp_path / ".muse"
76 for sub in ("refs/heads", "objects", "commits", "snapshots"):
77 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
78 (muse_dir / "repo.json").write_text(
79 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
80 )
81 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
82 (muse_dir / "refs" / "heads" / "main").write_text("")
83 (muse_dir / "config.toml").write_text("")
84 monkeypatch.chdir(tmp_path)
85 return tmp_path
86
87
88 def _json_get(result: InvokeResult) -> _GetJson:
89 for line in result.output.splitlines():
90 stripped = line.strip()
91 if stripped.startswith("{"):
92 d: _GetJson = json.loads(stripped)
93 return d
94 raise ValueError(f"No JSON in output:\n{result.output!r}")
95
96
97 def _json_set(result: InvokeResult) -> _SetJson:
98 for line in result.output.splitlines():
99 stripped = line.strip()
100 if stripped.startswith("{"):
101 d: _SetJson = json.loads(stripped)
102 return d
103 raise ValueError(f"No JSON in output:\n{result.output!r}")
104
105
106 def _json_show(result: InvokeResult) -> _ShowJson:
107 """Extract the first complete JSON object from potentially multi-line output."""
108 lines = result.output.splitlines()
109 start = None
110 for i, line in enumerate(lines):
111 if line.strip().startswith("{"):
112 start = i
113 break
114 if start is None:
115 raise ValueError(f"No JSON in output:\n{result.output!r}")
116 # Collect lines until braces balance
117 depth = 0
118 collected: list[str] = []
119 for line in lines[start:]:
120 collected.append(line)
121 depth += line.count("{") - line.count("}")
122 if depth <= 0:
123 break
124 d: _ShowJson = json.loads("\n".join(collected))
125 return d
126
127
128 # ── Unit: _escape ─────────────────────────────────────────────────────────────
129
130
131 class TestEscape:
132 def test_backslash_escaped(self) -> None:
133 from muse.cli.config import _escape
134 assert _escape("back\\slash") == "back\\\\slash"
135
136 def test_double_quote_escaped(self) -> None:
137 from muse.cli.config import _escape
138 assert _escape('say "hi"') == 'say \\"hi\\"'
139
140 def test_newline_escaped(self) -> None:
141 from muse.cli.config import _escape
142 assert _escape("line1\nline2") == "line1\\nline2"
143
144 def test_carriage_return_escaped(self) -> None:
145 from muse.cli.config import _escape
146 assert _escape("text\rmore") == "text\\rmore"
147
148 def test_null_byte_removed(self) -> None:
149 from muse.cli.config import _escape
150 assert "\0" not in _escape("bad\0byte")
151
152 def test_clean_string_passthrough(self) -> None:
153 from muse.cli.config import _escape
154 assert _escape("hello world") == "hello world"
155
156
157 # ── Unit: _validate_toml_key ──────────────────────────────────────────────────
158
159
160 class TestValidateTomlKey:
161 def test_newline_in_key_rejected(self) -> None:
162 from muse.cli.config import _validate_toml_key
163 with pytest.raises(ValueError, match="TOML keys"):
164 _validate_toml_key("bad\nkey")
165
166 def test_carriage_return_rejected(self) -> None:
167 from muse.cli.config import _validate_toml_key
168 with pytest.raises(ValueError, match="TOML keys"):
169 _validate_toml_key("bad\rkey")
170
171 def test_closing_bracket_rejected(self) -> None:
172 from muse.cli.config import _validate_toml_key
173 with pytest.raises(ValueError, match="TOML keys"):
174 _validate_toml_key("x]injection")
175
176 def test_opening_bracket_rejected(self) -> None:
177 from muse.cli.config import _validate_toml_key
178 with pytest.raises(ValueError, match="TOML keys"):
179 _validate_toml_key("[evil")
180
181 def test_equals_rejected(self) -> None:
182 from muse.cli.config import _validate_toml_key
183 with pytest.raises(ValueError, match="TOML keys"):
184 _validate_toml_key("k=v")
185
186 def test_double_quote_rejected(self) -> None:
187 from muse.cli.config import _validate_toml_key
188 with pytest.raises(ValueError, match="TOML keys"):
189 _validate_toml_key('key"val')
190
191 def test_null_byte_rejected(self) -> None:
192 from muse.cli.config import _validate_toml_key
193 with pytest.raises(ValueError, match="TOML keys"):
194 _validate_toml_key("bad\0key")
195
196 def test_safe_key_passes(self) -> None:
197 from muse.cli.config import _validate_toml_key
198 _validate_toml_key("ticks_per_beat")
199 _validate_toml_key("my-key.123")
200 _validate_toml_key("CamelCase")
201
202
203 # ── Unit: _dump_toml ──────────────────────────────────────────────────────────
204
205
206 class TestDumpToml:
207 def test_limits_section_round_trips(self) -> None:
208 from muse.cli.config import LimitsConfig, MuseConfig, _dump_toml
209 cfg: MuseConfig = {"limits": LimitsConfig(max_walk_commits=99, max_ancestors=500)}
210 toml_text = _dump_toml(cfg)
211 parsed = tomllib.loads(toml_text)
212 assert parsed["limits"]["max_walk_commits"] == 99
213 assert parsed["limits"]["max_ancestors"] == 500
214
215 def test_limits_shard_prefix_length_written(self) -> None:
216 from muse.cli.config import LimitsConfig, MuseConfig, _dump_toml
217 cfg: MuseConfig = {"limits": LimitsConfig(shard_prefix_length=4)}
218 toml_text = _dump_toml(cfg)
219 parsed = tomllib.loads(toml_text)
220 assert parsed["limits"]["shard_prefix_length"] == 4
221
222 def test_domain_key_injection_blocked_in_dump(self) -> None:
223 from muse.cli.config import MuseConfig, _dump_toml
224 cfg: MuseConfig = {"domain": {"evil\nkey": "val"}}
225 with pytest.raises(ValueError, match="TOML keys"):
226 _dump_toml(cfg)
227
228 def test_remote_name_injection_blocked(self) -> None:
229 from muse.cli.config import MuseConfig, RemoteEntry, _dump_toml
230 cfg: MuseConfig = {"remotes": {"evil\nname": RemoteEntry(url="http://localhost")}}
231 with pytest.raises(ValueError, match="TOML keys"):
232 _dump_toml(cfg)
233
234 def test_value_with_newline_escaped_not_injected(self) -> None:
235 from muse.cli.config import MuseConfig, _dump_toml
236 cfg: MuseConfig = {"domain": {"key": "line1\nline2"}}
237 toml_text = _dump_toml(cfg)
238 parsed = tomllib.loads(toml_text)
239 # The value line should contain \\n (escaped), not a literal newline
240 value_line = next(l for l in toml_text.splitlines() if l.startswith("key"))
241 assert "\n" not in value_line
242 assert "\\n" in value_line
243 assert "line1" in parsed["domain"]["key"]
244
245 def test_section_order(self) -> None:
246 from muse.cli.config import HubConfig, MuseConfig, UserConfig, _dump_toml
247 cfg: MuseConfig = {
248 "user": UserConfig(name="alice"),
249 "hub": HubConfig(url="https://musehub.ai"),
250 "domain": {"k": "v"},
251 }
252 toml_text = _dump_toml(cfg)
253 user_pos = toml_text.index("[user]")
254 hub_pos = toml_text.index("[hub]")
255 domain_pos = toml_text.index("[domain]")
256 assert user_pos < hub_pos < domain_pos
257
258
259 # ── Unit: set_config_value + get_config_value ─────────────────────────────────
260
261
262 class TestSetGetConfigValue:
263 def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
264 muse_dir = tmp_path / ".muse"
265 muse_dir.mkdir()
266 (muse_dir / "config.toml").write_text("")
267 return tmp_path
268
269 def test_limits_max_walk_commits_writes_to_limits_section(
270 self, tmp_path: pathlib.Path
271 ) -> None:
272 root = self._make_repo(tmp_path)
273 from muse.cli.config import set_config_value
274 set_config_value("limits.max_walk_commits", "5000", root)
275 raw = (root / ".muse" / "config.toml").read_text()
276 parsed = tomllib.loads(raw)
277 assert "limits" in parsed
278 assert parsed["limits"]["max_walk_commits"] == 5000
279 assert "domain" not in parsed
280
281 def test_limits_max_walk_commits_NOT_written_to_domain(
282 self, tmp_path: pathlib.Path
283 ) -> None:
284 """Regression: previously limits fell through to domain code path."""
285 root = self._make_repo(tmp_path)
286 from muse.cli.config import set_config_value
287 set_config_value("limits.max_walk_commits", "1000", root)
288 raw = (root / ".muse" / "config.toml").read_text()
289 parsed = tomllib.loads(raw)
290 assert "domain" not in parsed
291
292 def test_limits_shard_prefix_length_valid(self, tmp_path: pathlib.Path) -> None:
293 root = self._make_repo(tmp_path)
294 from muse.cli.config import get_config_value, set_config_value
295 set_config_value("limits.shard_prefix_length", "4", root)
296 assert get_config_value("limits.shard_prefix_length", root) == "4"
297
298 def test_limits_shard_prefix_length_invalid_rejected(
299 self, tmp_path: pathlib.Path
300 ) -> None:
301 root = self._make_repo(tmp_path)
302 from muse.cli.config import set_config_value
303 with pytest.raises(ValueError, match="shard_prefix_length must be 2 or 4"):
304 set_config_value("limits.shard_prefix_length", "3", root)
305
306 def test_limits_non_integer_rejected(self, tmp_path: pathlib.Path) -> None:
307 root = self._make_repo(tmp_path)
308 from muse.cli.config import set_config_value
309 with pytest.raises(ValueError, match="integer"):
310 set_config_value("limits.max_walk_commits", "notanint", root)
311
312 def test_limits_zero_rejected(self, tmp_path: pathlib.Path) -> None:
313 root = self._make_repo(tmp_path)
314 from muse.cli.config import set_config_value
315 with pytest.raises(ValueError, match="positive"):
316 set_config_value("limits.max_walk_commits", "0", root)
317
318 def test_limits_negative_rejected(self, tmp_path: pathlib.Path) -> None:
319 root = self._make_repo(tmp_path)
320 from muse.cli.config import set_config_value
321 with pytest.raises(ValueError, match="positive"):
322 set_config_value("limits.max_walk_commits", "-1", root)
323
324 def test_limits_unknown_key_rejected(self, tmp_path: pathlib.Path) -> None:
325 root = self._make_repo(tmp_path)
326 from muse.cli.config import set_config_value
327 with pytest.raises(ValueError, match="Unknown \\[limits\\]"):
328 set_config_value("limits.unknown_key", "5", root)
329
330 def test_domain_key_injection_rejected(self, tmp_path: pathlib.Path) -> None:
331 root = self._make_repo(tmp_path)
332 from muse.cli.config import set_config_value
333 with pytest.raises(ValueError, match="TOML keys"):
334 set_config_value("domain.evil\nkey", "bad", root)
335
336 def test_domain_key_with_bracket_injection_rejected(
337 self, tmp_path: pathlib.Path
338 ) -> None:
339 root = self._make_repo(tmp_path)
340 from muse.cli.config import set_config_value
341 with pytest.raises(ValueError, match="TOML keys"):
342 set_config_value("domain.x][evil", "bad", root)
343
344 def test_domain_safe_key_written(self, tmp_path: pathlib.Path) -> None:
345 root = self._make_repo(tmp_path)
346 from muse.cli.config import get_config_value, set_config_value
347 set_config_value("domain.ticks_per_beat", "480", root)
348 assert get_config_value("domain.ticks_per_beat", root) == "480"
349
350 def test_get_config_value_limits_after_write(self, tmp_path: pathlib.Path) -> None:
351 root = self._make_repo(tmp_path)
352 from muse.cli.config import get_config_value, set_config_value
353 set_config_value("limits.max_ancestors", "25000", root)
354 assert get_config_value("limits.max_ancestors", root) == "25000"
355
356
357 # ── Unit: config_as_dict ─────────────────────────────────────────────────────
358
359
360 class TestConfigAsDict:
361 def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
362 muse_dir = tmp_path / ".muse"
363 muse_dir.mkdir()
364 return tmp_path
365
366 def test_limits_included_in_output(self, tmp_path: pathlib.Path) -> None:
367 root = self._make_repo(tmp_path)
368 (root / ".muse" / "config.toml").write_text(
369 "[limits]\nmax_walk_commits = 5000\n"
370 )
371 from muse.cli.config import config_as_dict
372 d = config_as_dict(root)
373 assert "limits" in d
374 assert d["limits"]["max_walk_commits"] == "5000"
375
376 def test_limits_absent_when_not_set(self, tmp_path: pathlib.Path) -> None:
377 root = self._make_repo(tmp_path)
378 (root / ".muse" / "config.toml").write_text("[user]\nname = \"alice\"\n")
379 from muse.cli.config import config_as_dict
380 d = config_as_dict(root)
381 assert "limits" not in d
382
383 def test_empty_config_returns_empty_dict(self, tmp_path: pathlib.Path) -> None:
384 root = self._make_repo(tmp_path)
385 (root / ".muse" / "config.toml").write_text("")
386 from muse.cli.config import config_as_dict
387 assert config_as_dict(root) == {}
388
389
390 # ── Integration: run_show ─────────────────────────────────────────────────────
391
392
393 class TestRunShow:
394 def test_show_json_includes_limits(self, repo: pathlib.Path) -> None:
395 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "7777"])
396 result = runner.invoke(cli, ["config", "show", "--json"])
397 assert result.exit_code == 0
398 data = _json_show(result)
399 assert "limits" in data
400 limits = data["limits"]
401 assert isinstance(limits, dict)
402 assert limits["max_walk_commits"] == "7777"
403
404 def test_show_json_schema_user(self, repo: pathlib.Path) -> None:
405 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
406 result = runner.invoke(cli, ["config", "show", "--json"])
407 assert result.exit_code == 0
408 data = _json_show(result)
409 user = data.get("user")
410 assert isinstance(user, dict)
411 assert user.get("name") == "Alice"
412
413 def test_show_format_json_alias(self, repo: pathlib.Path) -> None:
414 result = runner.invoke(cli, ["config", "show", "--format", "json"])
415 assert result.exit_code == 0
416 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
417 assert len(json_lines) >= 1
418
419 def test_show_format_invalid_exits(self, repo: pathlib.Path) -> None:
420 result = runner.invoke(cli, ["config", "show", "--format", "xml"])
421 assert result.exit_code != 0
422
423 def test_show_text_mode_ansi_sanitized(
424 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
425 ) -> None:
426 # Write a config with ANSI in value (directly to file to bypass validation)
427 (repo / ".muse" / "config.toml").write_text(
428 '[user]\nname = "\\x1b[31mevil\\x1b[0m"\n'
429 )
430 result = runner.invoke(cli, ["config", "show"])
431 assert "\x1b[" not in result.output
432
433 def test_show_text_empty_config(self, repo: pathlib.Path) -> None:
434 result = runner.invoke(cli, ["config", "show"])
435 assert result.exit_code == 0
436 assert "No configuration set" in result.output
437
438 def test_show_text_limits_section_displayed(self, repo: pathlib.Path) -> None:
439 runner.invoke(cli, ["config", "set", "limits.max_ancestors", "30000"])
440 result = runner.invoke(cli, ["config", "show"])
441 assert result.exit_code == 0
442 assert "[limits]" in result.output
443 assert "max_ancestors" in result.output
444
445 def test_show_json_stdout_clean(self, repo: pathlib.Path) -> None:
446 runner.invoke(cli, ["config", "set", "user.name", "Bob"])
447 result = runner.invoke(cli, ["config", "show", "--json"])
448 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
449 assert len(json_lines) >= 1
450
451 def test_show_no_repo_still_works(
452 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
453 ) -> None:
454 # show gracefully shows empty config outside a repo (uses cwd)
455 monkeypatch.chdir(tmp_path)
456 result = runner.invoke(cli, ["config", "show"])
457 assert result.exit_code == 0
458
459
460 # ── Integration: run_get ─────────────────────────────────────────────────────
461
462
463 class TestRunGet:
464 def test_get_existing_key_raw_value(self, repo: pathlib.Path) -> None:
465 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
466 result = runner.invoke(cli, ["config", "get", "user.name"])
467 assert result.exit_code == 0
468 assert "Alice" in result.output
469
470 def test_get_json_schema(self, repo: pathlib.Path) -> None:
471 runner.invoke(cli, ["config", "set", "user.type", "agent"])
472 result = runner.invoke(cli, ["config", "get", "user.type", "--json"])
473 assert result.exit_code == 0
474 data = _json_get(result)
475 assert data["key"] == "user.type"
476 assert data["value"] == "agent"
477
478 def test_get_missing_key_exits_nonzero(self, repo: pathlib.Path) -> None:
479 result = runner.invoke(cli, ["config", "get", "user.name"])
480 assert result.exit_code != 0
481
482 def test_get_missing_key_error_to_stderr(
483 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
484 ) -> None:
485 result = runner.invoke(cli, ["config", "get", "user.name"])
486 assert result.exit_code != 0
487 assert "not set" in result.output
488
489 def test_get_limits_key_after_set(self, repo: pathlib.Path) -> None:
490 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "12345"])
491 result = runner.invoke(cli, ["config", "get", "limits.max_walk_commits"])
492 assert result.exit_code == 0
493 assert "12345" in result.output
494
495 def test_get_json_stdout_clean(self, repo: pathlib.Path) -> None:
496 runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
497 result = runner.invoke(cli, ["config", "get", "user.email", "--json"])
498 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
499 assert len(json_lines) >= 1
500
501
502 # ── Integration: run_set ─────────────────────────────────────────────────────
503
504
505 class TestRunSet:
506 def test_set_success_json_schema(self, repo: pathlib.Path) -> None:
507 result = runner.invoke(
508 cli, ["config", "set", "user.name", "Bob", "--json"]
509 )
510 assert result.exit_code == 0
511 data = _json_set(result)
512 assert data["status"] == "ok"
513 assert data["key"] == "user.name"
514 assert data["value"] == "Bob"
515
516 def test_set_success_stderr_message(
517 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
518 ) -> None:
519 result = runner.invoke(cli, ["config", "set", "user.name", "Carol"])
520 assert result.exit_code == 0
521 assert "Carol" in result.output
522
523 def test_set_json_stdout_clean(self, repo: pathlib.Path) -> None:
524 result = runner.invoke(
525 cli, ["config", "set", "user.type", "agent", "--json"]
526 )
527 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
528 assert len(json_lines) >= 1
529
530 def test_set_blocked_namespace_exits(self, repo: pathlib.Path) -> None:
531 result = runner.invoke(cli, ["config", "set", "auth.token", "secret"])
532 assert result.exit_code != 0
533
534 def test_set_blocked_remotes_namespace_exits(self, repo: pathlib.Path) -> None:
535 result = runner.invoke(cli, ["config", "set", "remotes.origin", "url"])
536 assert result.exit_code != 0
537
538 def test_set_domain_newline_injection_rejected(self, repo: pathlib.Path) -> None:
539 result = runner.invoke(cli, ["config", "set", "domain.evil\nkey", "bad"])
540 assert result.exit_code != 0
541
542 def test_set_domain_bracket_injection_rejected(self, repo: pathlib.Path) -> None:
543 result = runner.invoke(cli, ["config", "set", "domain.x][evil", "bad"])
544 assert result.exit_code != 0
545
546 def test_set_limits_max_walk_commits(self, repo: pathlib.Path) -> None:
547 result = runner.invoke(
548 cli, ["config", "set", "limits.max_walk_commits", "20000"]
549 )
550 assert result.exit_code == 0
551 get_result = runner.invoke(cli, ["config", "get", "limits.max_walk_commits"])
552 assert "20000" in get_result.output
553
554 def test_set_limits_non_integer_rejected(self, repo: pathlib.Path) -> None:
555 result = runner.invoke(
556 cli, ["config", "set", "limits.max_walk_commits", "abc"]
557 )
558 assert result.exit_code != 0
559
560 def test_set_limits_zero_rejected(self, repo: pathlib.Path) -> None:
561 result = runner.invoke(
562 cli, ["config", "set", "limits.max_walk_commits", "0"]
563 )
564 assert result.exit_code != 0
565
566 def test_set_limits_shard_prefix_valid(self, repo: pathlib.Path) -> None:
567 result = runner.invoke(
568 cli, ["config", "set", "limits.shard_prefix_length", "4"]
569 )
570 assert result.exit_code == 0
571
572 def test_set_limits_shard_prefix_invalid_rejected(self, repo: pathlib.Path) -> None:
573 result = runner.invoke(
574 cli, ["config", "set", "limits.shard_prefix_length", "3"]
575 )
576 assert result.exit_code != 0
577
578 def test_set_error_ansi_sanitized_in_output(
579 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
580 ) -> None:
581 # Use a key with ANSI that would appear in the error message
582 ansi_key = "domain.\x1b[31mevil\x1b[0m\nkey"
583 result = runner.invoke(cli, ["config", "set", ansi_key, "val"])
584 assert result.exit_code != 0
585 assert "\x1b[" not in result.output
586
587 def test_set_limits_writes_to_limits_not_domain(self, repo: pathlib.Path) -> None:
588 """Regression: limits namespace must not fall through to domain code."""
589 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "9999"])
590 raw = (repo / ".muse" / "config.toml").read_text()
591 parsed = tomllib.loads(raw)
592 assert "domain" not in parsed
593 assert parsed["limits"]["max_walk_commits"] == 9999
594
595
596 # ── Integration: run_edit ─────────────────────────────────────────────────────
597
598
599 class TestRunEdit:
600 def test_edit_no_repo_exits(
601 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
602 ) -> None:
603 monkeypatch.chdir(tmp_path)
604 result = runner.invoke(cli, ["config", "edit"])
605 assert result.exit_code != 0
606
607 def test_edit_missing_config_file_autocreated(
608 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
609 ) -> None:
610 """Missing config.toml must be auto-created before the editor opens."""
611 (repo / ".muse" / "config.toml").unlink()
612 monkeypatch.setenv("EDITOR", "true")
613 monkeypatch.delenv("VISUAL", raising=False)
614 result = runner.invoke(cli, ["config", "edit"])
615 assert result.exit_code == 0
616 assert (repo / ".muse" / "config.toml").exists()
617
618 def test_edit_bad_editor_exits(
619 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
620 ) -> None:
621 monkeypatch.setenv("EDITOR", "nonexistent-editor-xyz")
622 monkeypatch.delenv("VISUAL", raising=False)
623 result = runner.invoke(cli, ["config", "edit"])
624 assert result.exit_code != 0
625
626 def test_edit_invokes_editor(
627 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
628 ) -> None:
629 monkeypatch.setenv("EDITOR", "true")
630 monkeypatch.delenv("VISUAL", raising=False)
631 result = runner.invoke(cli, ["config", "edit"])
632 assert result.exit_code == 0
633
634
635 # ── Security ──────────────────────────────────────────────────────────────────
636
637
638 class TestConfigSecurity:
639 def test_toml_key_injection_blocked_end_to_end(self, repo: pathlib.Path) -> None:
640 """Setting domain key with newline must not corrupt config.toml."""
641 result = runner.invoke(
642 cli, ["config", "set", "domain.evil\nkey", "bad"]
643 )
644 assert result.exit_code != 0
645 raw = (repo / ".muse" / "config.toml").read_text()
646 assert "\nkey" not in raw
647
648 def test_bracket_injection_blocked(self, repo: pathlib.Path) -> None:
649 result = runner.invoke(
650 cli, ["config", "set", "domain.x][evil", "val"]
651 )
652 assert result.exit_code != 0
653 raw = (repo / ".muse" / "config.toml").read_text()
654 assert "[evil]" not in raw
655
656 def test_equals_injection_blocked(self, repo: pathlib.Path) -> None:
657 result = runner.invoke(
658 cli, ["config", "set", "domain.x=y", "val"]
659 )
660 assert result.exit_code != 0
661
662 def test_ansi_in_show_text_stripped(self, repo: pathlib.Path) -> None:
663 (repo / ".muse" / "config.toml").write_text(
664 '[domain]\nticks = "\\x1b[31mred\\x1b[0m"\n'
665 )
666 result = runner.invoke(cli, ["config", "show"])
667 assert "\x1b[" not in result.output
668
669 def test_auth_namespace_always_blocked(self, repo: pathlib.Path) -> None:
670 for key in ("auth.token", "auth.password", "auth.secret"):
671 result = runner.invoke(cli, ["config", "set", key, "val"])
672 assert result.exit_code != 0, f"Expected {key!r} to be blocked"
673
674 def test_credentials_not_in_json_output(self, repo: pathlib.Path) -> None:
675 (repo / ".muse" / "config.toml").write_text(
676 '[hub]\nurl = "http://localhost:10003"\n'
677 '[auth]\ntoken = "secret-token"\n'
678 )
679 result = runner.invoke(cli, ["config", "show", "--json"])
680 assert result.exit_code == 0
681 assert "secret-token" not in result.output
682
683 def test_value_newline_escaped_in_toml(self, repo: pathlib.Path) -> None:
684 runner.invoke(cli, ["config", "set", "user.name", "Alice\nBob"])
685 raw = (repo / ".muse" / "config.toml").read_text()
686 # The literal newline must not appear in the value field
687 name_line = [l for l in raw.splitlines() if "name" in l][0]
688 assert "\n" not in name_line
689
690 def test_error_message_to_stderr_not_stdout(self, repo: pathlib.Path) -> None:
691 result = runner.invoke(
692 cli, ["config", "set", "domain.evil\nkey", "val"]
693 )
694 assert result.exit_code != 0
695 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
696 assert len(json_lines) == 0
697
698
699 # ── E2E round-trips ───────────────────────────────────────────────────────────
700
701
702 class TestE2ERoundTrips:
703 def test_set_then_get_user(self, repo: pathlib.Path) -> None:
704 runner.invoke(cli, ["config", "set", "user.name", "DeepBlue"])
705 result = runner.invoke(cli, ["config", "get", "user.name"])
706 assert result.exit_code == 0
707 assert "DeepBlue" in result.output
708
709 def test_set_limits_then_show_json(self, repo: pathlib.Path) -> None:
710 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "3333"])
711 result = runner.invoke(cli, ["config", "show", "--json"])
712 assert result.exit_code == 0
713 data = _json_show(result)
714 limits = data.get("limits")
715 assert isinstance(limits, dict)
716 assert limits.get("max_walk_commits") == "3333"
717
718 def test_set_domain_then_show_valid_toml(self, repo: pathlib.Path) -> None:
719 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
720 result = runner.invoke(cli, ["config", "show"])
721 assert result.exit_code == 0
722 assert "960" in result.output
723
724 def test_limits_written_to_limits_section_not_domain(
725 self, repo: pathlib.Path
726 ) -> None:
727 runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "8888"])
728 raw = (repo / ".muse" / "config.toml").read_text()
729 parsed = tomllib.loads(raw)
730 assert "domain" not in parsed
731 assert parsed["limits"]["max_graph_commits"] == 8888
732
733 def test_multiple_writes_preserve_all_sections(self, repo: pathlib.Path) -> None:
734 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
735 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
736 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "2000"])
737 raw = (repo / ".muse" / "config.toml").read_text()
738 parsed = tomllib.loads(raw)
739 assert parsed["user"]["name"] == "Alice"
740 assert parsed["domain"]["ticks_per_beat"] == "480"
741 assert parsed["limits"]["max_walk_commits"] == 2000
742
743 def test_set_json_get_json_consistent(self, repo: pathlib.Path) -> None:
744 set_result = runner.invoke(
745 cli, ["config", "set", "user.type", "agent", "--json"]
746 )
747 get_result = runner.invoke(
748 cli, ["config", "get", "user.type", "--json"]
749 )
750 set_data = _json_set(set_result)
751 get_data = _json_get(get_result)
752 assert set_data["value"] == get_data["value"] == "agent"
753
754
755 # ── Stress ────────────────────────────────────────────────────────────────────
756
757
758 class TestStress:
759 def test_8_concurrent_set_to_isolated_repos(
760 self, tmp_path: pathlib.Path
761 ) -> None:
762 """8 threads writing to independent repos must not corrupt each other."""
763 from muse._version import __version__
764 from muse.cli.config import get_config_value, set_config_value
765
766 errors: list[str] = []
767
768 def _do(idx: int) -> None:
769 try:
770 repo_dir = tmp_path / f"repo_{idx}"
771 muse_dir = repo_dir / ".muse"
772 muse_dir.mkdir(parents=True)
773 (muse_dir / "config.toml").write_text("")
774 (muse_dir / "repo.json").write_text(
775 json.dumps({
776 "repo_id": f"repo-{idx}",
777 "schema_version": __version__,
778 "domain": "code",
779 })
780 )
781 value = f"user_{idx}"
782 set_config_value("user.name", value, repo_dir)
783 result = get_config_value("user.name", repo_dir)
784 assert result == value, f"Expected {value!r}, got {result!r}"
785 except Exception as exc:
786 errors.append(f"Thread {idx}: {exc}")
787
788 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
789 for t in threads:
790 t.start()
791 for t in threads:
792 t.join()
793 assert errors == [], "Concurrent config write failures:\n" + "\n".join(errors)
794
795 def test_8_concurrent_set_limits_isolated(
796 self, tmp_path: pathlib.Path
797 ) -> None:
798 """8 threads writing limits to isolated repos must write to [limits] not [domain]."""
799 from muse._version import __version__
800 from muse.cli.config import set_config_value
801
802 errors: list[str] = []
803
804 def _do(idx: int) -> None:
805 try:
806 repo_dir = tmp_path / f"limits_repo_{idx}"
807 muse_dir = repo_dir / ".muse"
808 muse_dir.mkdir(parents=True)
809 (muse_dir / "config.toml").write_text("")
810 (muse_dir / "repo.json").write_text(
811 json.dumps({
812 "repo_id": f"repo-{idx}",
813 "schema_version": __version__,
814 "domain": "code",
815 })
816 )
817 set_config_value("limits.max_walk_commits", str(1000 + idx), repo_dir)
818 raw = (muse_dir / "config.toml").read_text()
819 parsed = tomllib.loads(raw)
820 assert "limits" in parsed, "limits section missing"
821 assert "domain" not in parsed, "limits fell through to domain"
822 assert parsed["limits"]["max_walk_commits"] == 1000 + idx
823 except Exception as exc:
824 errors.append(f"Thread {idx}: {exc}")
825
826 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
827 for t in threads:
828 t.start()
829 for t in threads:
830 t.join()
831 assert errors == [], "Concurrent limits write failures:\n" + "\n".join(errors)
832
833
834 # =============================================================================
835 # muse config show — extended hardening
836 # =============================================================================
837
838
839 class TestRunShowExtended:
840 """Additional coverage for ``muse config show`` gaps."""
841
842 # ── flag aliases ──────────────────────────────────────────────────────────
843
844 def test_j_short_flag_emits_json(self, repo: pathlib.Path) -> None:
845 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
846 result = runner.invoke(cli, ["config", "show", "-j"])
847 assert result.exit_code == 0
848 data = _json_show(result)
849 assert isinstance(data, dict)
850
851 def test_j_flag_same_as_json_flag(self, repo: pathlib.Path) -> None:
852 runner.invoke(cli, ["config", "set", "user.name", "Bob"])
853 r1 = runner.invoke(cli, ["config", "show", "--json"])
854 r2 = runner.invoke(cli, ["config", "show", "-j"])
855 assert r1.exit_code == 0
856 assert r2.exit_code == 0
857 assert _json_show(r1) == _json_show(r2)
858
859 def test_f_short_flag_json_alias(self, repo: pathlib.Path) -> None:
860 runner.invoke(cli, ["config", "set", "user.name", "Carol"])
861 result = runner.invoke(cli, ["config", "show", "-f", "json"])
862 assert result.exit_code == 0
863 data = _json_show(result)
864 assert isinstance(data, dict)
865
866 def test_format_text_explicit(self, repo: pathlib.Path) -> None:
867 runner.invoke(cli, ["config", "set", "user.name", "Dave"])
868 result = runner.invoke(cli, ["config", "show", "--format", "text"])
869 assert result.exit_code == 0
870 assert "[user]" in result.output
871 assert "{" not in result.output # no JSON
872
873 def test_json_and_format_json_together_no_double_output(
874 self, repo: pathlib.Path
875 ) -> None:
876 """--json and --format json together must not emit duplicate output."""
877 runner.invoke(cli, ["config", "set", "user.name", "Eve"])
878 result = runner.invoke(cli, ["config", "show", "--json", "--format", "json"])
879 assert result.exit_code == 0
880 # Must be parseable as a single JSON object, not two concatenated objects
881 data = _json_show(result)
882 assert isinstance(data, dict)
883
884 # ── JSON structure ────────────────────────────────────────────────────────
885
886 def test_json_is_pretty_printed(self, repo: pathlib.Path) -> None:
887 """JSON output must use indent=2 — agents need readable diffs."""
888 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
889 result = runner.invoke(cli, ["config", "show", "--json"])
890 assert result.exit_code == 0
891 # Pretty-printed JSON has multiple lines
892 lines = [l for l in result.output.splitlines() if l.strip()]
893 assert len(lines) > 1
894
895 def test_json_hub_section_present(self, repo: pathlib.Path) -> None:
896 from muse.cli.config import set_hub_url
897 set_hub_url("https://musehub.ai", repo)
898 result = runner.invoke(cli, ["config", "show", "--json"])
899 assert result.exit_code == 0
900 data = _json_show(result)
901 assert "hub" in data
902 hub = data["hub"]
903 assert isinstance(hub, dict)
904 assert hub["url"] == "https://musehub.ai"
905
906 def test_json_remotes_section_present(self, repo: pathlib.Path) -> None:
907 from muse.cli.config import set_remote
908 set_remote("origin", "https://hub.example.com/owner/repo", repo)
909 result = runner.invoke(cli, ["config", "show", "--json"])
910 assert result.exit_code == 0
911 data = _json_show(result)
912 assert "remotes" in data
913 remotes = data["remotes"]
914 assert isinstance(remotes, dict)
915 assert "origin" in remotes
916
917 def test_json_domain_section_present(self, repo: pathlib.Path) -> None:
918 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
919 result = runner.invoke(cli, ["config", "show", "--json"])
920 assert result.exit_code == 0
921 data = _json_show(result)
922 assert "domain" in data
923 domain = data["domain"]
924 assert isinstance(domain, dict)
925 assert domain["ticks_per_beat"] == "960"
926
927 def test_json_all_sections_together(self, repo: pathlib.Path) -> None:
928 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
929 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
930 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "5000"])
931 result = runner.invoke(cli, ["config", "show", "--json"])
932 assert result.exit_code == 0
933 data = _json_show(result)
934 assert "user" in data
935 assert "domain" in data
936 assert "limits" in data
937
938 def test_json_empty_config_is_empty_object(self, repo: pathlib.Path) -> None:
939 result = runner.invoke(cli, ["config", "show", "--json"])
940 assert result.exit_code == 0
941 data = _json_show(result)
942 assert data == {}
943
944 def test_json_no_credentials(self, repo: pathlib.Path) -> None:
945 """Credentials (auth, token) must never appear in JSON output."""
946 # Write a config with a fake [auth] section directly
947 (repo / ".muse" / "config.toml").write_text(
948 '[user]\nname = "Alice"\n\n[auth]\ntoken = "secret"\n'
949 )
950 result = runner.invoke(cli, ["config", "show", "--json"])
951 assert result.exit_code == 0
952 assert "secret" not in result.output
953 assert "token" not in result.output
954
955 # ── text mode structure ───────────────────────────────────────────────────
956
957 def test_text_output_to_stdout(self, repo: pathlib.Path) -> None:
958 """Text mode config content goes to stdout, not only stderr."""
959 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
960 result = runner.invoke(cli, ["config", "show"])
961 assert result.exit_code == 0
962 assert "[user]" in result.output
963
964 def test_text_user_section_header(self, repo: pathlib.Path) -> None:
965 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
966 result = runner.invoke(cli, ["config", "show"])
967 assert "[user]" in result.output
968
969 def test_text_hub_section_header(self, repo: pathlib.Path) -> None:
970 from muse.cli.config import set_hub_url
971 set_hub_url("https://musehub.ai", repo)
972 result = runner.invoke(cli, ["config", "show"])
973 assert "[hub]" in result.output
974
975 def test_text_remotes_section_header(self, repo: pathlib.Path) -> None:
976 from muse.cli.config import set_remote
977 set_remote("origin", "https://hub.example.com/owner/repo", repo)
978 result = runner.invoke(cli, ["config", "show"])
979 assert "[remotes." in result.output
980
981 def test_text_domain_section_header(self, repo: pathlib.Path) -> None:
982 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
983 result = runner.invoke(cli, ["config", "show"])
984 assert "[domain]" in result.output
985
986 def test_text_key_value_format(self, repo: pathlib.Path) -> None:
987 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
988 result = runner.invoke(cli, ["config", "show"])
989 # TOML key = "value" format
990 assert 'name = "Alice"' in result.output
991
992 def test_text_limits_no_quotes_on_values(self, repo: pathlib.Path) -> None:
993 """Limits values are integers in TOML — no quotes."""
994 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "8000"])
995 result = runner.invoke(cli, ["config", "show"])
996 assert "max_walk_commits = 8000" in result.output
997
998 def test_show_help_contains_quickstart(self) -> None:
999 result = runner.invoke(cli, ["config", "show", "--help"])
1000 assert result.exit_code == 0
1001 assert "quickstart" in result.output.lower() or "jq" in result.output
1002
1003 def test_show_help_contains_exit_codes(self) -> None:
1004 result = runner.invoke(cli, ["config", "show", "--help"])
1005 assert result.exit_code == 0
1006 assert "Exit codes" in result.output or "exit" in result.output.lower()
1007
1008
1009 class TestRunShowStress:
1010 """Stress tests for ``muse config show``."""
1011
1012 def test_concurrent_show_reads(self, repo: pathlib.Path) -> None:
1013 """Concurrent config_as_dict reads of the same config must all succeed.
1014
1015 Uses config_as_dict directly — CliRunner shares a buffer across threads
1016 so full CLI invocations cannot be called concurrently from different threads.
1017 """
1018 import threading
1019 from muse.cli.config import config_as_dict, set_config_value
1020 set_config_value("user.name", "Alice", repo)
1021 set_config_value("limits.max_walk_commits", "1000", repo)
1022 errors: list[str] = []
1023
1024 def _do(idx: int) -> None:
1025 try:
1026 data = config_as_dict(repo)
1027 assert "user" in data
1028 assert data["user"]["name"] == "Alice"
1029 except Exception as exc:
1030 errors.append(f"Thread {idx}: {exc}")
1031
1032 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1033 for t in threads:
1034 t.start()
1035 for t in threads:
1036 t.join()
1037 assert errors == [], "\n".join(errors)
1038
1039 def test_show_large_domain_config(self, repo: pathlib.Path) -> None:
1040 """Show handles a config with many domain keys without truncation."""
1041 for i in range(20):
1042 runner.invoke(cli, ["config", "set", f"domain.key_{i}", str(i * 10)])
1043 result = runner.invoke(cli, ["config", "show", "--json"])
1044 assert result.exit_code == 0
1045 data = _json_show(result)
1046 domain = data.get("domain")
1047 assert isinstance(domain, dict)
1048 assert len(domain) == 20
1049
1050 def test_show_json_output_is_valid_json(self, repo: pathlib.Path) -> None:
1051 """JSON output must always be parseable regardless of config contents."""
1052 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
1053 runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
1054 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1055 runner.invoke(cli, ["config", "set", "limits.max_ancestors", "50000"])
1056 result = runner.invoke(cli, ["config", "show", "--json"])
1057 assert result.exit_code == 0
1058 # json.loads raises on invalid JSON
1059 parsed = json.loads(result.output)
1060 assert isinstance(parsed, dict)
1061
1062
1063 # =============================================================================
1064 # muse config get — extended hardening
1065 # =============================================================================
1066
1067
1068 class TestRunGetExtended:
1069 """Additional coverage for ``muse config get`` gaps."""
1070
1071 # ── flag aliases ──────────────────────────────────────────────────────────
1072
1073 def test_j_short_flag_emits_json(self, repo: pathlib.Path) -> None:
1074 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
1075 result = runner.invoke(cli, ["config", "get", "user.name", "-j"])
1076 assert result.exit_code == 0
1077 data = _json_get(result)
1078 assert data["value"] == "Alice"
1079
1080 def test_j_same_as_json_flag(self, repo: pathlib.Path) -> None:
1081 runner.invoke(cli, ["config", "set", "user.name", "Bob"])
1082 r1 = runner.invoke(cli, ["config", "get", "user.name", "--json"])
1083 r2 = runner.invoke(cli, ["config", "get", "user.name", "-j"])
1084 assert r1.exit_code == 0 and r2.exit_code == 0
1085 assert _json_get(r1) == _json_get(r2)
1086
1087 # ── key format validation ─────────────────────────────────────────────────
1088
1089 def test_key_without_dot_exits_nonzero(self, repo: pathlib.Path) -> None:
1090 result = runner.invoke(cli, ["config", "get", "username"])
1091 assert result.exit_code != 0
1092
1093 def test_key_without_dot_shows_format_hint(self, repo: pathlib.Path) -> None:
1094 result = runner.invoke(cli, ["config", "get", "username"])
1095 assert "namespace.subkey" in result.output or "user.name" in result.output
1096
1097 def test_key_without_dot_does_not_say_not_set(
1098 self, repo: pathlib.Path
1099 ) -> None:
1100 """Malformed key must not produce the misleading 'is not set' message."""
1101 result = runner.invoke(cli, ["config", "get", "badkey"])
1102 assert "is not set" not in result.output
1103
1104 def test_empty_key_exits_nonzero(self, repo: pathlib.Path) -> None:
1105 result = runner.invoke(cli, ["config", "get", ""])
1106 assert result.exit_code != 0
1107
1108 def test_unknown_namespace_exits_nonzero(self, repo: pathlib.Path) -> None:
1109 result = runner.invoke(cli, ["config", "get", "unknown.key"])
1110 assert result.exit_code != 0
1111
1112 # ── all supported key namespaces ──────────────────────────────────────────
1113
1114 def test_get_user_email(self, repo: pathlib.Path) -> None:
1115 runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
1116 result = runner.invoke(cli, ["config", "get", "user.email"])
1117 assert result.exit_code == 0
1118 assert "[email protected]" in result.output
1119
1120 def test_get_user_type(self, repo: pathlib.Path) -> None:
1121 runner.invoke(cli, ["config", "set", "user.type", "agent"])
1122 result = runner.invoke(cli, ["config", "get", "user.type"])
1123 assert result.exit_code == 0
1124 assert "agent" in result.output
1125
1126 def test_get_hub_url(self, repo: pathlib.Path) -> None:
1127 from muse.cli.config import set_hub_url
1128 set_hub_url("https://musehub.ai", repo)
1129 result = runner.invoke(cli, ["config", "get", "hub.url"])
1130 assert result.exit_code == 0
1131 assert "musehub.ai" in result.output
1132
1133 def test_get_domain_key(self, repo: pathlib.Path) -> None:
1134 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
1135 result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"])
1136 assert result.exit_code == 0
1137 assert "960" in result.output
1138
1139 def test_get_limits_max_ancestors(self, repo: pathlib.Path) -> None:
1140 runner.invoke(cli, ["config", "set", "limits.max_ancestors", "20000"])
1141 result = runner.invoke(cli, ["config", "get", "limits.max_ancestors"])
1142 assert result.exit_code == 0
1143 assert "20000" in result.output
1144
1145 def test_get_limits_max_graph_commits(self, repo: pathlib.Path) -> None:
1146 runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "500"])
1147 result = runner.invoke(cli, ["config", "get", "limits.max_graph_commits"])
1148 assert result.exit_code == 0
1149 assert "500" in result.output
1150
1151 def test_get_limits_shard_prefix_length(self, repo: pathlib.Path) -> None:
1152 runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "4"])
1153 result = runner.invoke(cli, ["config", "get", "limits.shard_prefix_length"])
1154 assert result.exit_code == 0
1155 assert "4" in result.output
1156
1157 # ── JSON structure ────────────────────────────────────────────────────────
1158
1159 def test_json_key_field_matches_input(self, repo: pathlib.Path) -> None:
1160 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
1161 result = runner.invoke(cli, ["config", "get", "user.name", "--json"])
1162 assert result.exit_code == 0
1163 data = _json_get(result)
1164 assert data["key"] == "user.name"
1165
1166 def test_json_value_matches_text_output(self, repo: pathlib.Path) -> None:
1167 runner.invoke(cli, ["config", "set", "user.name", "Charlie"])
1168 r_text = runner.invoke(cli, ["config", "get", "user.name"])
1169 r_json = runner.invoke(cli, ["config", "get", "user.name", "--json"])
1170 assert r_text.exit_code == 0 and r_json.exit_code == 0
1171 json_value = _json_get(r_json)["value"]
1172 assert json_value in r_text.output
1173
1174 def test_json_missing_key_exits_nonzero(self, repo: pathlib.Path) -> None:
1175 result = runner.invoke(cli, ["config", "get", "user.name", "--json"])
1176 assert result.exit_code != 0
1177
1178 def test_json_stdout_clean_on_success(self, repo: pathlib.Path) -> None:
1179 runner.invoke(cli, ["config", "set", "user.name", "Dave"])
1180 result = runner.invoke(cli, ["config", "get", "user.name", "--json"])
1181 assert result.exit_code == 0
1182 # Output must be valid JSON
1183 parsed = json.loads(result.output)
1184 assert "key" in parsed and "value" in parsed
1185
1186 # ── text mode output ──────────────────────────────────────────────────────
1187
1188 def test_text_value_printed_to_stdout(self, repo: pathlib.Path) -> None:
1189 runner.invoke(cli, ["config", "set", "user.name", "Eve"])
1190 result = runner.invoke(cli, ["config", "get", "user.name"])
1191 assert result.exit_code == 0
1192 assert "Eve" in result.output
1193
1194 def test_text_error_goes_to_stderr_marker(self, repo: pathlib.Path) -> None:
1195 """Not-set error must not appear in stdout (goes to stderr)."""
1196 result = runner.invoke(cli, ["config", "get", "user.name"])
1197 # CliRunner merges stderr — ensure exit is nonzero and message present
1198 assert result.exit_code != 0
1199 assert "not set" in result.output or "not" in result.output.lower()
1200
1201 def test_get_help_contains_key_reference(self) -> None:
1202 result = runner.invoke(cli, ["config", "get", "--help"])
1203 assert result.exit_code == 0
1204 assert "user.name" in result.output or "hub.url" in result.output
1205
1206 def test_get_help_contains_exit_codes(self) -> None:
1207 result = runner.invoke(cli, ["config", "get", "--help"])
1208 assert result.exit_code == 0
1209 assert "Exit codes" in result.output or "exit" in result.output.lower()
1210
1211 # ── outside repo ─────────────────────────────────────────────────────────
1212
1213 def test_get_outside_repo_key_not_set(
1214 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1215 ) -> None:
1216 """Outside a repo, all keys return not-set (gracefully)."""
1217 monkeypatch.chdir(tmp_path)
1218 result = runner.invoke(cli, ["config", "get", "user.name"])
1219 assert result.exit_code != 0
1220
1221
1222 class TestRunGetStress:
1223 """Stress tests for ``muse config get``."""
1224
1225 def test_concurrent_get_reads(self, repo: pathlib.Path) -> None:
1226 """Concurrent config_as_dict reads are thread-safe."""
1227 import threading
1228 from muse.cli.config import get_config_value, set_config_value
1229 set_config_value("user.name", "Stress", repo)
1230 errors: list[str] = []
1231
1232 def _do(idx: int) -> None:
1233 try:
1234 value = get_config_value("user.name", repo)
1235 assert value == "Stress"
1236 except Exception as exc:
1237 errors.append(f"Thread {idx}: {exc}")
1238
1239 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1240 for t in threads:
1241 t.start()
1242 for t in threads:
1243 t.join()
1244 assert errors == [], "\n".join(errors)
1245
1246 def test_all_limits_keys_readable(self, repo: pathlib.Path) -> None:
1247 """All four limits keys must be individually gettable after set."""
1248 pairs = [
1249 ("limits.max_walk_commits", "9000"),
1250 ("limits.max_ancestors", "40000"),
1251 ("limits.max_graph_commits", "250"),
1252 ("limits.shard_prefix_length", "4"),
1253 ]
1254 for key, val in pairs:
1255 runner.invoke(cli, ["config", "set", key, val])
1256 for key, expected in pairs:
1257 result = runner.invoke(cli, ["config", "get", key])
1258 assert result.exit_code == 0, f"Failed for {key}"
1259 assert expected in result.output, f"Expected {expected!r} for {key}"
1260
1261 def test_get_many_different_keys_sequentially(
1262 self, repo: pathlib.Path
1263 ) -> None:
1264 """Reading many keys in sequence must not corrupt state."""
1265 runner.invoke(cli, ["config", "set", "user.name", "Alice"])
1266 runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
1267 runner.invoke(cli, ["config", "set", "user.type", "human"])
1268 runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1269 runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "1000"])
1270 for _ in range(20):
1271 for key, expected in [
1272 ("user.name", "Alice"),
1273 ("user.email", "[email protected]"),
1274 ("user.type", "human"),
1275 ("domain.ticks_per_beat", "480"),
1276 ("limits.max_walk_commits", "1000"),
1277 ]:
1278 result = runner.invoke(cli, ["config", "get", key])
1279 assert result.exit_code == 0
1280 assert expected in result.output
1281
1282
1283 # ── Extended: run_set ─────────────────────────────────────────────────────────
1284
1285
1286 class TestRunSetExtended:
1287 """Extended hardening tests for ``muse config set``."""
1288
1289 def test_j_alias_works(self, repo: pathlib.Path) -> None:
1290 result = runner.invoke(cli, ["config", "set", "user.name", "Alice", "-j"])
1291 assert result.exit_code == 0
1292 data = _json_set(result)
1293 assert data["status"] == "ok"
1294 assert data["key"] == "user.name"
1295 assert data["value"] == "Alice"
1296
1297 def test_key_without_dot_exits_with_format_error(self, repo: pathlib.Path) -> None:
1298 result = runner.invoke(cli, ["config", "set", "username", "Alice"])
1299 assert result.exit_code != 0
1300 assert "namespace.subkey" in result.output
1301
1302 def test_key_without_dot_mentions_example(self, repo: pathlib.Path) -> None:
1303 result = runner.invoke(cli, ["config", "set", "badkey", "val"])
1304 assert result.exit_code != 0
1305 assert "user.name" in result.output or "hub.url" in result.output
1306
1307 def test_unknown_namespace_exits_nonzero(self, repo: pathlib.Path) -> None:
1308 result = runner.invoke(cli, ["config", "set", "mystery.key", "val"])
1309 assert result.exit_code != 0
1310
1311 def test_unknown_namespace_error_message_helpful(self, repo: pathlib.Path) -> None:
1312 result = runner.invoke(cli, ["config", "set", "mystery.key", "val"])
1313 assert "mystery" in result.output or "Unknown" in result.output
1314
1315 def test_user_name_settable(self, repo: pathlib.Path) -> None:
1316 result = runner.invoke(cli, ["config", "set", "user.name", "Bob"])
1317 assert result.exit_code == 0
1318
1319 def test_user_email_settable(self, repo: pathlib.Path) -> None:
1320 result = runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
1321 assert result.exit_code == 0
1322
1323 def test_user_type_human_settable(self, repo: pathlib.Path) -> None:
1324 result = runner.invoke(cli, ["config", "set", "user.type", "human"])
1325 assert result.exit_code == 0
1326
1327 def test_user_type_agent_settable(self, repo: pathlib.Path) -> None:
1328 result = runner.invoke(cli, ["config", "set", "user.type", "agent"])
1329 assert result.exit_code == 0
1330
1331 def test_hub_url_https_settable(self, repo: pathlib.Path) -> None:
1332 result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1333 assert result.exit_code == 0
1334
1335 def test_hub_url_http_rejected(self, repo: pathlib.Path) -> None:
1336 result = runner.invoke(cli, ["config", "set", "hub.url", "http://musehub.ai"])
1337 assert result.exit_code != 0
1338
1339 def test_hub_unknown_subkey_rejected(self, repo: pathlib.Path) -> None:
1340 result = runner.invoke(cli, ["config", "set", "hub.secret", "val"])
1341 assert result.exit_code != 0
1342
1343 def test_domain_key_settable(self, repo: pathlib.Path) -> None:
1344 result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1345 assert result.exit_code == 0
1346
1347 def test_limits_max_ancestors_settable(self, repo: pathlib.Path) -> None:
1348 result = runner.invoke(cli, ["config", "set", "limits.max_ancestors", "25000"])
1349 assert result.exit_code == 0
1350 get_result = runner.invoke(cli, ["config", "get", "limits.max_ancestors"])
1351 assert "25000" in get_result.output
1352
1353 def test_limits_max_graph_commits_settable(self, repo: pathlib.Path) -> None:
1354 result = runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "5000"])
1355 assert result.exit_code == 0
1356 get_result = runner.invoke(cli, ["config", "get", "limits.max_graph_commits"])
1357 assert "5000" in get_result.output
1358
1359 def test_limits_shard_prefix_2_valid(self, repo: pathlib.Path) -> None:
1360 result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "2"])
1361 assert result.exit_code == 0
1362
1363 def test_limits_shard_prefix_4_valid(self, repo: pathlib.Path) -> None:
1364 result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "4"])
1365 assert result.exit_code == 0
1366
1367 def test_limits_shard_prefix_1_rejected(self, repo: pathlib.Path) -> None:
1368 result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "1"])
1369 assert result.exit_code != 0
1370
1371 def test_limits_shard_prefix_3_rejected(self, repo: pathlib.Path) -> None:
1372 result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "3"])
1373 assert result.exit_code != 0
1374
1375 def test_limits_negative_rejected(self, repo: pathlib.Path) -> None:
1376 result = runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "-5"])
1377 assert result.exit_code != 0
1378
1379 def test_blocked_auth_shows_redirect(self, repo: pathlib.Path) -> None:
1380 result = runner.invoke(cli, ["config", "set", "auth.token", "secret"])
1381 assert result.exit_code != 0
1382 assert "auth" in result.output.lower() or "login" in result.output.lower()
1383
1384 def test_blocked_remotes_shows_redirect(self, repo: pathlib.Path) -> None:
1385 result = runner.invoke(cli, ["config", "set", "remotes.origin", "url"])
1386 assert result.exit_code != 0
1387 assert "remote" in result.output.lower()
1388
1389 def test_success_json_status_ok(self, repo: pathlib.Path) -> None:
1390 result = runner.invoke(cli, ["config", "set", "user.name", "X", "--json"])
1391 data = _json_set(result)
1392 assert data["status"] == "ok"
1393
1394 def test_success_json_stdout_only_has_json(self, repo: pathlib.Path) -> None:
1395 result = runner.invoke(cli, ["config", "set", "user.name", "Y", "--json"])
1396 assert result.exit_code == 0
1397 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1398 assert len(json_lines) >= 1
1399
1400 def test_success_text_goes_to_output(self, repo: pathlib.Path) -> None:
1401 result = runner.invoke(cli, ["config", "set", "user.name", "Carol"])
1402 assert result.exit_code == 0
1403 assert "Carol" in result.output
1404
1405 def test_help_contains_settable_namespaces(self, repo: pathlib.Path) -> None:
1406 result = runner.invoke(cli, ["config", "set", "--help"])
1407 assert "user.name" in result.output
1408 assert "hub.url" in result.output
1409 assert "domain" in result.output
1410 assert "limits" in result.output
1411
1412 def test_help_contains_blocked_namespaces(self, repo: pathlib.Path) -> None:
1413 result = runner.invoke(cli, ["config", "set", "--help"])
1414 assert "auth" in result.output
1415 assert "remotes" in result.output
1416
1417 def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1418 result = runner.invoke(cli, ["config", "set", "--help"])
1419 assert "Exit" in result.output or "exit" in result.output
1420
1421
1422 # ── Security: run_set ─────────────────────────────────────────────────────────
1423
1424
1425 class TestRunSetSecurity:
1426 """Security-focused tests for ``muse config set``."""
1427
1428 def test_ansi_in_key_sanitized_in_error(self, repo: pathlib.Path) -> None:
1429 result = runner.invoke(cli, ["config", "set", "domain.\x1b[31mevil\x1b[0m\nkey", "val"])
1430 assert result.exit_code != 0
1431 assert "\x1b[" not in result.output
1432
1433 def test_ansi_in_value_sanitized_in_text_success(self, repo: pathlib.Path) -> None:
1434 result = runner.invoke(cli, ["config", "set", "user.name", "\x1b[31mBob\x1b[0m"])
1435 assert result.exit_code == 0
1436 assert "\x1b[" not in result.output
1437
1438 def test_null_byte_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1439 result = runner.invoke(cli, ["config", "set", "domain.evil\x00key", "val"])
1440 assert result.exit_code != 0
1441
1442 def test_bracket_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1443 result = runner.invoke(cli, ["config", "set", "domain.x][evil", "val"])
1444 assert result.exit_code != 0
1445
1446 def test_equals_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1447 result = runner.invoke(cli, ["config", "set", "domain.key=evil", "val"])
1448 assert result.exit_code != 0
1449
1450 def test_newline_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1451 result = runner.invoke(cli, ["config", "set", "domain.evil\nkey", "val"])
1452 assert result.exit_code != 0
1453
1454 def test_key_without_dot_shows_format_hint_not_generic_error(self, repo: pathlib.Path) -> None:
1455 """Ensures format error, not a generic 'unknown namespace' message."""
1456 result = runner.invoke(cli, ["config", "set", "nodot", "val"])
1457 assert result.exit_code != 0
1458 assert "namespace.subkey" in result.output
1459
1460 def test_hub_http_blocked_not_stored(self, repo: pathlib.Path) -> None:
1461 runner.invoke(cli, ["config", "set", "hub.url", "http://evil.example.com"])
1462 result = runner.invoke(cli, ["config", "get", "hub.url"])
1463 assert result.exit_code != 0 or "evil.example.com" not in result.output
1464
1465
1466 # ── Stress: run_set ───────────────────────────────────────────────────────────
1467
1468
1469 class TestRunSetStress:
1470 """Concurrency and volume tests for ``muse config set``."""
1471
1472 def test_concurrent_set_to_different_repos_no_corruption(
1473 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1474 ) -> None:
1475 """8 threads writing to separate repos must each produce correct values."""
1476 import threading
1477 from muse._version import __version__
1478 from muse.cli.config import get_config_value, set_config_value
1479
1480 errors: list[str] = []
1481
1482 def _worker(idx: int) -> None:
1483 repo_path = tmp_path / f"repo_{idx}"
1484 muse_dir = repo_path / ".muse"
1485 for sub in ("refs/heads", "objects", "commits", "snapshots"):
1486 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
1487 (muse_dir / "repo.json").write_text(
1488 json.dumps({"repo_id": f"repo-{idx}", "schema_version": __version__, "domain": "code"})
1489 )
1490 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
1491 (muse_dir / "config.toml").write_text("")
1492 expected = f"user_{idx}"
1493 set_config_value("user.name", expected, repo_path)
1494 got = get_config_value("user.name", repo_path)
1495 if got != expected:
1496 errors.append(f"repo_{idx}: expected {expected!r}, got {got!r}")
1497
1498 threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
1499 for t in threads:
1500 t.start()
1501 for t in threads:
1502 t.join()
1503 assert errors == [], "\n".join(errors)
1504
1505 def test_all_four_limits_keys_set_round_trip(self, repo: pathlib.Path) -> None:
1506 """All four limits keys written then read back."""
1507 from muse.cli.config import get_config_value, set_config_value
1508
1509 pairs = [
1510 ("limits.max_walk_commits", "10000"),
1511 ("limits.max_ancestors", "5000"),
1512 ("limits.max_graph_commits", "2500"),
1513 ("limits.shard_prefix_length", "4"),
1514 ]
1515 for key, val in pairs:
1516 set_config_value(key, val, repo)
1517 for key, expected in pairs:
1518 got = get_config_value(key, repo)
1519 assert got == expected, f"{key}: expected {expected!r}, got {got!r}"
1520
1521 def test_20_sequential_sets_no_state_corruption(self, repo: pathlib.Path) -> None:
1522 """Writing 20 different user.name values sequentially — last write wins."""
1523 from muse.cli.config import get_config_value, set_config_value
1524
1525 for i in range(20):
1526 set_config_value("user.name", f"user_{i}", repo)
1527 got = get_config_value("user.name", repo)
1528 assert got == "user_19"
1529
1530
1531 # ── Extended: run_edit ────────────────────────────────────────────────────────
1532
1533
1534 class TestRunEditExtended:
1535 """Extended hardening tests for ``muse config edit``."""
1536
1537 def test_visual_takes_precedence_over_editor(
1538 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1539 ) -> None:
1540 """$VISUAL must be used when both $VISUAL and $EDITOR are set."""
1541 calls: list[list[str]] = []
1542
1543 import subprocess as _sp
1544
1545 real_run = _sp.run
1546
1547 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1548 calls.append(cmd)
1549 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1550
1551 monkeypatch.setattr(_sp, "run", _fake_run)
1552 monkeypatch.setenv("VISUAL", "visual-editor")
1553 monkeypatch.setenv("EDITOR", "editor-fallback")
1554 runner.invoke(cli, ["config", "edit"])
1555 assert calls and calls[0][0] == "visual-editor"
1556
1557 def test_editor_used_when_visual_unset(
1558 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1559 ) -> None:
1560 calls: list[list[str]] = []
1561
1562 import subprocess as _sp
1563
1564 real_run = _sp.run
1565
1566 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1567 calls.append(cmd)
1568 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1569
1570 monkeypatch.setattr(_sp, "run", _fake_run)
1571 monkeypatch.delenv("VISUAL", raising=False)
1572 monkeypatch.setenv("EDITOR", "my-editor")
1573 runner.invoke(cli, ["config", "edit"])
1574 assert calls and calls[0][0] == "my-editor"
1575
1576 def test_vi_fallback_when_both_unset(
1577 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1578 ) -> None:
1579 calls: list[list[str]] = []
1580
1581 import subprocess as _sp
1582
1583 real_run = _sp.run
1584
1585 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1586 calls.append(cmd)
1587 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1588
1589 monkeypatch.setattr(_sp, "run", _fake_run)
1590 monkeypatch.delenv("VISUAL", raising=False)
1591 monkeypatch.delenv("EDITOR", raising=False)
1592 runner.invoke(cli, ["config", "edit"])
1593 assert calls and calls[0][0] == "vi"
1594
1595 def test_multiword_editor_split_correctly(
1596 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1597 ) -> None:
1598 """EDITOR='code --wait' must be split to ['code', '--wait', path]."""
1599 calls: list[list[str]] = []
1600
1601 import subprocess as _sp
1602
1603 real_run = _sp.run
1604
1605 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1606 calls.append(cmd)
1607 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1608
1609 monkeypatch.setattr(_sp, "run", _fake_run)
1610 monkeypatch.setenv("EDITOR", "code --wait")
1611 monkeypatch.delenv("VISUAL", raising=False)
1612 runner.invoke(cli, ["config", "edit"])
1613 assert calls and calls[0][:2] == ["code", "--wait"]
1614
1615 def test_multiword_editor_passes_config_path(
1616 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1617 ) -> None:
1618 """Multi-word editor: config path must be the final argument."""
1619 calls: list[list[str]] = []
1620
1621 import subprocess as _sp
1622
1623 real_run = _sp.run
1624
1625 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1626 calls.append(cmd)
1627 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1628
1629 monkeypatch.setattr(_sp, "run", _fake_run)
1630 monkeypatch.setenv("EDITOR", "emacs -nw")
1631 monkeypatch.delenv("VISUAL", raising=False)
1632 runner.invoke(cli, ["config", "edit"])
1633 assert calls and "config.toml" in calls[0][-1]
1634
1635 def test_auto_create_config_when_missing(
1636 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1637 ) -> None:
1638 (repo / ".muse" / "config.toml").unlink()
1639 monkeypatch.setenv("EDITOR", "true")
1640 monkeypatch.delenv("VISUAL", raising=False)
1641 assert not (repo / ".muse" / "config.toml").exists()
1642 result = runner.invoke(cli, ["config", "edit"])
1643 assert result.exit_code == 0
1644 assert (repo / ".muse" / "config.toml").exists()
1645
1646 def test_auto_create_info_message_on_stderr(
1647 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1648 ) -> None:
1649 (repo / ".muse" / "config.toml").unlink()
1650 monkeypatch.setenv("EDITOR", "true")
1651 monkeypatch.delenv("VISUAL", raising=False)
1652 result = runner.invoke(cli, ["config", "edit"])
1653 assert "Created" in result.output or "config.toml" in result.output
1654
1655 def test_editor_invoked_as_list_not_shell(
1656 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1657 ) -> None:
1658 """subprocess.run must receive a list, never shell=True."""
1659 captured: MsgpackDict = {}
1660
1661 import subprocess as _sp
1662
1663 def _fake_run(cmd: list[str] | str, **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1664 captured["cmd"] = cmd
1665 captured["shell"] = kwargs.get("shell", False)
1666 from subprocess import CompletedProcess
1667 return CompletedProcess([], 0)
1668
1669 monkeypatch.setattr(_sp, "run", _fake_run)
1670 monkeypatch.setenv("EDITOR", "true")
1671 monkeypatch.delenv("VISUAL", raising=False)
1672 runner.invoke(cli, ["config", "edit"])
1673 assert isinstance(captured.get("cmd"), list)
1674 assert not captured.get("shell")
1675
1676 def test_editor_nonzero_exit_exits_nonzero(
1677 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1678 ) -> None:
1679 monkeypatch.setenv("EDITOR", "false") # /bin/false always exits 1
1680 monkeypatch.delenv("VISUAL", raising=False)
1681 result = runner.invoke(cli, ["config", "edit"])
1682 assert result.exit_code != 0
1683
1684 def test_editor_nonzero_exit_message_contains_code(
1685 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1686 ) -> None:
1687 monkeypatch.setenv("EDITOR", "false")
1688 monkeypatch.delenv("VISUAL", raising=False)
1689 result = runner.invoke(cli, ["config", "edit"])
1690 assert result.exit_code != 0
1691 # Message should mention the exit code number
1692 assert any(ch.isdigit() for ch in result.output)
1693
1694 def test_outside_repo_exits_nonzero(
1695 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1696 ) -> None:
1697 monkeypatch.chdir(tmp_path)
1698 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1699 result = runner.invoke(cli, ["config", "edit"])
1700 assert result.exit_code != 0
1701
1702 def test_outside_repo_error_message(
1703 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1704 ) -> None:
1705 monkeypatch.chdir(tmp_path)
1706 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1707 result = runner.invoke(cli, ["config", "edit"])
1708 assert "repository" in result.output.lower() or "repo" in result.output.lower()
1709
1710 def test_help_mentions_visual(self, repo: pathlib.Path) -> None:
1711 result = runner.invoke(cli, ["config", "edit", "--help"])
1712 assert "VISUAL" in result.output
1713
1714 def test_help_mentions_editor(self, repo: pathlib.Path) -> None:
1715 result = runner.invoke(cli, ["config", "edit", "--help"])
1716 assert "EDITOR" in result.output
1717
1718 def test_help_mentions_agent_alternative(self, repo: pathlib.Path) -> None:
1719 result = runner.invoke(cli, ["config", "edit", "--help"])
1720 assert "muse config set" in result.output or "agent" in result.output
1721
1722 def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1723 result = runner.invoke(cli, ["config", "edit", "--help"])
1724 assert "Exit" in result.output or "exit" in result.output
1725
1726 def test_config_path_passed_to_editor(
1727 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1728 ) -> None:
1729 """Editor must receive the config.toml path as its argument."""
1730 calls: list[list[str]] = []
1731
1732 import subprocess as _sp
1733
1734 real_run = _sp.run
1735
1736 def _fake_run(cmd: list[str], **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1737 calls.append(cmd)
1738 return real_run(["true"], **{k: v for k, v in kwargs.items()})
1739
1740 monkeypatch.setattr(_sp, "run", _fake_run)
1741 monkeypatch.setenv("EDITOR", "my-editor")
1742 monkeypatch.delenv("VISUAL", raising=False)
1743 runner.invoke(cli, ["config", "edit"])
1744 assert calls
1745 assert str(repo / ".muse" / "config.toml") in calls[0]
1746
1747
1748 # ── Security: run_edit ────────────────────────────────────────────────────────
1749
1750
1751 class TestRunEditSecurity:
1752 """Security-focused tests for ``muse config edit``."""
1753
1754 def test_ansi_in_editor_env_sanitized_in_error(
1755 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1756 ) -> None:
1757 monkeypatch.setenv("EDITOR", "\x1b[31mevil\x1b[0m-editor")
1758 monkeypatch.delenv("VISUAL", raising=False)
1759 result = runner.invoke(cli, ["config", "edit"])
1760 assert result.exit_code != 0
1761 assert "\x1b[" not in result.output
1762
1763 def test_editor_invoked_without_shell(
1764 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1765 ) -> None:
1766 """Verify shell=True is never passed to subprocess.run."""
1767 captured: MsgpackDict = {}
1768
1769 import subprocess as _sp
1770
1771 def _fake_run(cmd: list[str] | str, **kwargs: str | bool | int | None) -> _sp.CompletedProcess[bytes]:
1772 captured["shell"] = kwargs.get("shell", False)
1773 from subprocess import CompletedProcess
1774 return CompletedProcess([], 0)
1775
1776 monkeypatch.setattr(_sp, "run", _fake_run)
1777 monkeypatch.setenv("EDITOR", "true")
1778 monkeypatch.delenv("VISUAL", raising=False)
1779 runner.invoke(cli, ["config", "edit"])
1780 assert not captured.get("shell")
1781
1782 def test_malformed_editor_command_exits_gracefully(
1783 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1784 ) -> None:
1785 """An unparseable $EDITOR value must exit cleanly, not crash."""
1786 # shlex.split raises ValueError on unmatched quotes
1787 monkeypatch.setenv("EDITOR", "editor 'unclosed quote")
1788 monkeypatch.delenv("VISUAL", raising=False)
1789 result = runner.invoke(cli, ["config", "edit"])
1790 assert result.exit_code != 0
1791
1792 def test_shell_metacharacters_in_editor_not_shell_executed(
1793 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1794 ) -> None:
1795 """$EDITOR with shell metacharacters must not trigger shell execution.
1796
1797 shlex.split turns 'true; echo injected' into ['true;', 'echo', 'injected'].
1798 subprocess.run receives a list (never shell=True), so it tries to exec a
1799 binary literally named 'true;' — which doesn't exist. FileNotFoundError
1800 fires; no shell command is ever evaluated.
1801 """
1802 monkeypatch.setenv("EDITOR", "true; echo injected")
1803 monkeypatch.delenv("VISUAL", raising=False)
1804 result = runner.invoke(cli, ["config", "edit"])
1805 # Binary 'true;' doesn't exist → non-zero exit; no shell evaluation.
1806 assert result.exit_code != 0
1807 # Error message quotes the editor string, not the result of shell execution.
1808 assert "Editor not found" in result.output
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago