gabriel / muse public
test_cmd_init.py python
1,586 lines 62.2 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
1 """Comprehensive tests for ``muse init``.
2
3 Coverage tiers:
4 - Unit: template generators, _copy_template, module constants
5 - CLI unit: argument validation (bad branch, bad domain, edge cases)
6 - Integration: every flag, file format, lifecycle scenario, file preservation
7 - End-to-end: muse init followed by subsequent muse commands
8 - Security: ANSI injection, symlink attacks, TOCTOU, corrupt inputs
9 - Stress: sequential, concurrent, large-scale, adversarial inputs
10 """
11 from __future__ import annotations
12
13 import json
14 import os
15 import pathlib
16 import threading
17 import tomllib
18 import uuid
19
20 import pytest
21
22 from tests.cli_test_helper import CliRunner, InvokeResult
23
24 runner = CliRunner()
25
26
27 def _init(repo: pathlib.Path, *extra_args: str) -> InvokeResult:
28 """Invoke ``muse init`` inside *repo* (created if needed)."""
29 from muse.cli.app import main as cli
30
31 repo.mkdir(parents=True, exist_ok=True)
32 saved = os.getcwd()
33 try:
34 os.chdir(repo)
35 return runner.invoke(cli, ["init", *extra_args])
36 finally:
37 os.chdir(saved)
38
39
40 # ---------------------------------------------------------------------------
41 # Unit — template generators
42 # ---------------------------------------------------------------------------
43
44
45 class TestMuseignoreTemplate:
46 def test_code_domain_has_pyc_pattern(self) -> None:
47 from muse.cli.commands.init import _museignore_template
48
49 result = _museignore_template("code")
50 assert "*.pyc" in result
51
52 def test_midi_domain_has_renders_pattern(self) -> None:
53 from muse.cli.commands.init import _museignore_template
54
55 result = _museignore_template("midi")
56 assert "/renders/" in result
57
58 def test_unknown_domain_produces_commented_stub(self) -> None:
59 from muse.cli.commands.init import _museignore_template
60
61 result = _museignore_template("genomics")
62 assert "[domain.genomics]" in result
63 assert "# patterns" in result
64
65 def test_result_is_valid_toml(self) -> None:
66 from muse.cli.commands.init import _museignore_template
67
68 for domain in ("code", "midi", "genomics"):
69 parsed = tomllib.loads(_museignore_template(domain))
70 assert isinstance(parsed, dict)
71
72 def test_global_section_present(self) -> None:
73 from muse.cli.commands.init import _museignore_template
74
75 parsed = tomllib.loads(_museignore_template("code"))
76 assert "global" in parsed
77 assert isinstance(parsed["global"]["patterns"], list)
78 assert ".DS_Store" in parsed["global"]["patterns"]
79
80 def test_code_domain_ignores_tls_keys(self) -> None:
81 """*.key files (TLS private keys) must be excluded by default in code repos."""
82 from muse.cli.commands.init import _museignore_template
83
84 result = _museignore_template("code")
85 assert "*.key" in result
86
87 def test_code_domain_ignores_tls_certs(self) -> None:
88 """*.crt and *.pem files must be excluded by default in code repos."""
89 from muse.cli.commands.init import _museignore_template
90
91 result = _museignore_template("code")
92 assert "*.crt" in result
93 assert "*.pem" in result
94
95 def test_tls_patterns_are_valid_toml(self) -> None:
96 """Template with TLS patterns must still parse as valid TOML."""
97 from muse.cli.commands.init import _museignore_template
98
99 parsed = tomllib.loads(_museignore_template("code"))
100 code_patterns = parsed.get("domain", {}).get("code", {}).get("patterns", [])
101 assert any(p == "*.key" for p in code_patterns)
102 assert any(p == "*.crt" for p in code_patterns)
103 assert any(p == "*.pem" for p in code_patterns)
104
105
106 class TestMuseattributesTemplate:
107 def test_domain_embedded_in_meta(self) -> None:
108 from muse.cli.commands.init import _museattributes_template
109
110 result = _museattributes_template("code")
111 parsed = tomllib.loads(result)
112 assert parsed["meta"]["domain"] == "code"
113
114 def test_custom_domain_embedded(self) -> None:
115 from muse.cli.commands.init import _museattributes_template
116
117 result = _museattributes_template("genomics")
118 parsed = tomllib.loads(result)
119 assert parsed["meta"]["domain"] == "genomics"
120
121 def test_result_is_valid_toml(self) -> None:
122 from muse.cli.commands.init import _museattributes_template
123
124 for domain in ("code", "midi", "genomics"):
125 parsed = tomllib.loads(_museattributes_template(domain))
126 assert isinstance(parsed, dict)
127
128
129 # ---------------------------------------------------------------------------
130 # CLI unit — argument validation
131 # ---------------------------------------------------------------------------
132
133
134 class TestArgValidation:
135 def test_invalid_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
136 # Null byte is explicitly forbidden by validate_branch_name
137 result = _init(tmp_path, "--default-branch", "branch\x00null")
138 assert result.exit_code != 0
139
140 def test_invalid_branch_name_json_error(self, tmp_path: pathlib.Path) -> None:
141 # Consecutive dots are forbidden (path traversal prevention)
142 result = _init(tmp_path, "--default-branch", "../traversal", "--json")
143 assert result.exit_code != 0
144 data = json.loads(result.output)
145 assert "error" in data
146
147 def test_invalid_domain_name_rejected(self, tmp_path: pathlib.Path) -> None:
148 # Domain names must match [a-z][a-z0-9_-]* — spaces and ! are banned
149 result = _init(tmp_path, "--domain", "Bad-Domain!")
150 assert result.exit_code != 0
151
152 def test_invalid_domain_name_json_error(self, tmp_path: pathlib.Path) -> None:
153 result = _init(tmp_path, "--domain", "Bad-Domain!", "--json")
154 assert result.exit_code != 0
155 data = json.loads(result.output)
156 assert "error" in data
157
158 def test_missing_template_dir_rejected(self, tmp_path: pathlib.Path) -> None:
159 result = _init(tmp_path, "--template", str(tmp_path / "nonexistent"))
160 assert result.exit_code != 0
161
162 def test_missing_template_dir_json_error(self, tmp_path: pathlib.Path) -> None:
163 result = _init(tmp_path, "--template", str(tmp_path / "nonexistent"), "--json")
164 assert result.exit_code != 0
165 data = json.loads(result.output)
166 assert "error" in data
167
168 def test_template_pointing_to_file_rejected(self, tmp_path: pathlib.Path) -> None:
169 f = tmp_path / "not_a_dir.txt"
170 f.write_text("hello")
171 result = _init(tmp_path / "repo", "--template", str(f))
172 assert result.exit_code != 0
173
174 def test_reinit_without_force_rejected(self, tmp_path: pathlib.Path) -> None:
175 _init(tmp_path)
176 result = _init(tmp_path)
177 assert result.exit_code != 0
178
179 def test_reinit_without_force_json_error(self, tmp_path: pathlib.Path) -> None:
180 _init(tmp_path)
181 result = _init(tmp_path, "--json")
182 assert result.exit_code != 0
183 data = json.loads(result.output)
184 assert "error" in data
185
186
187 # ---------------------------------------------------------------------------
188 # Integration — filesystem layout
189 # ---------------------------------------------------------------------------
190
191
192 class TestFilesystemLayout:
193 def test_muse_dir_created(self, tmp_path: pathlib.Path) -> None:
194 assert _init(tmp_path).exit_code == 0
195 assert (tmp_path / ".muse").is_dir()
196
197 def test_required_subdirs_exist(self, tmp_path: pathlib.Path) -> None:
198 _init(tmp_path)
199 muse = tmp_path / ".muse"
200 for subdir in ("objects", "commits", "snapshots", "refs", "refs/heads"):
201 assert (muse / subdir).is_dir(), f"missing: {subdir}"
202
203 def test_repo_json_created(self, tmp_path: pathlib.Path) -> None:
204 _init(tmp_path)
205 assert (tmp_path / ".muse" / "repo.json").exists()
206
207 def test_repo_json_fields(self, tmp_path: pathlib.Path) -> None:
208 _init(tmp_path)
209 data = json.loads((tmp_path / ".muse" / "repo.json").read_text())
210 assert "repo_id" in data
211 assert "schema_version" in data
212 assert "created_at" in data
213 assert "domain" in data
214 assert data["domain"] == "code"
215
216 def test_repo_json_repo_id_is_sha256(self, tmp_path: pathlib.Path) -> None:
217 _init(tmp_path)
218 raw = json.loads((tmp_path / ".muse" / "repo.json").read_text())["repo_id"]
219 assert raw.startswith("sha256:"), f"expected sha256: prefix, got {raw!r}"
220 assert len(raw) == 71
221
222 def test_head_points_to_default_branch(self, tmp_path: pathlib.Path) -> None:
223 _init(tmp_path)
224 head = (tmp_path / ".muse" / "HEAD").read_text()
225 assert "main" in head
226
227 def test_custom_default_branch_in_head(self, tmp_path: pathlib.Path) -> None:
228 _init(tmp_path, "--default-branch", "dev")
229 head = (tmp_path / ".muse" / "HEAD").read_text()
230 assert "dev" in head
231 assert (tmp_path / ".muse" / "refs" / "heads" / "dev").exists()
232
233 def test_config_toml_created(self, tmp_path: pathlib.Path) -> None:
234 _init(tmp_path)
235 assert (tmp_path / ".muse" / "config.toml").exists()
236
237 def test_museignore_created(self, tmp_path: pathlib.Path) -> None:
238 _init(tmp_path)
239 assert (tmp_path / ".museignore").exists()
240
241 def test_museattributes_created(self, tmp_path: pathlib.Path) -> None:
242 _init(tmp_path)
243 assert (tmp_path / ".museattributes").exists()
244
245 def test_museattributes_has_correct_domain(self, tmp_path: pathlib.Path) -> None:
246 _init(tmp_path, "--domain", "code")
247 parsed = tomllib.loads((tmp_path / ".museattributes").read_text())
248 assert parsed["meta"]["domain"] == "code"
249
250 def test_museignore_valid_toml(self, tmp_path: pathlib.Path) -> None:
251 _init(tmp_path)
252 parsed = tomllib.loads((tmp_path / ".museignore").read_text())
253 assert isinstance(parsed, dict)
254
255
256 class TestBareRepo:
257 def test_bare_creates_muse_dir(self, tmp_path: pathlib.Path) -> None:
258 assert _init(tmp_path, "--bare").exit_code == 0
259 assert (tmp_path / ".muse").is_dir()
260
261 def test_bare_does_not_create_museignore(self, tmp_path: pathlib.Path) -> None:
262 _init(tmp_path, "--bare")
263 assert not (tmp_path / ".museignore").exists()
264
265 def test_bare_does_not_create_museattributes(self, tmp_path: pathlib.Path) -> None:
266 _init(tmp_path, "--bare")
267 assert not (tmp_path / ".museattributes").exists()
268
269 def test_bare_repo_json_has_bare_flag(self, tmp_path: pathlib.Path) -> None:
270 _init(tmp_path, "--bare")
271 data = json.loads((tmp_path / ".muse" / "repo.json").read_text())
272 assert data.get("bare") is True
273
274 def test_non_bare_repo_json_has_no_bare_flag(self, tmp_path: pathlib.Path) -> None:
275 _init(tmp_path)
276 data = json.loads((tmp_path / ".muse" / "repo.json").read_text())
277 assert "bare" not in data
278
279
280 class TestForceReinit:
281 def test_force_reinit_succeeds(self, tmp_path: pathlib.Path) -> None:
282 _init(tmp_path)
283 result = _init(tmp_path, "--force")
284 assert result.exit_code == 0
285
286 def test_force_preserves_repo_id(self, tmp_path: pathlib.Path) -> None:
287 _init(tmp_path)
288 original_id = json.loads(
289 (tmp_path / ".muse" / "repo.json").read_text()
290 )["repo_id"]
291 _init(tmp_path, "--force")
292 new_id = json.loads(
293 (tmp_path / ".muse" / "repo.json").read_text()
294 )["repo_id"]
295 assert original_id == new_id
296
297 def test_force_does_not_overwrite_museignore(self, tmp_path: pathlib.Path) -> None:
298 _init(tmp_path)
299 custom = '[global]\npatterns = ["custom.txt"]\n'
300 (tmp_path / ".museignore").write_text(custom)
301 _init(tmp_path, "--force")
302 assert (tmp_path / ".museignore").read_text() == custom
303
304 def test_force_does_not_overwrite_museattributes(self, tmp_path: pathlib.Path) -> None:
305 _init(tmp_path)
306 custom = '[meta]\ndomain = "custom"\n'
307 (tmp_path / ".museattributes").write_text(custom)
308 _init(tmp_path, "--force")
309 assert (tmp_path / ".museattributes").read_text() == custom
310
311 def test_force_on_fresh_dir_works(self, tmp_path: pathlib.Path) -> None:
312 # --force on a directory that was never a repo should work as fresh init
313 result = _init(tmp_path, "--force")
314 assert result.exit_code == 0
315 assert (tmp_path / ".muse" / "repo.json").exists()
316
317
318 class TestTemplate:
319 def test_template_files_copied(self, tmp_path: pathlib.Path) -> None:
320 tmpl = tmp_path / "tmpl"
321 tmpl.mkdir()
322 (tmpl / "README.md").write_text("# hello")
323 (tmpl / "scripts").mkdir()
324 (tmpl / "scripts" / "run.sh").write_text("#!/bin/sh\necho hi")
325
326 repo = tmp_path / "repo"
327 repo.mkdir()
328 _init(repo, "--template", str(tmpl))
329
330 assert (repo / "README.md").read_text() == "# hello"
331 assert (repo / "scripts" / "run.sh").exists()
332
333 def test_template_does_not_overwrite_muse_dir(self, tmp_path: pathlib.Path) -> None:
334 tmpl = tmp_path / "tmpl"
335 tmpl.mkdir()
336 # A template that tries to write a .muse directory
337 (tmpl / ".muse").mkdir()
338 (tmpl / ".muse" / "injected").write_text("bad")
339
340 repo = tmp_path / "repo"
341 repo.mkdir()
342 _init(repo, "--template", str(tmpl))
343
344 # The injected file may land but should not corrupt the real repo.json
345 assert (repo / ".muse" / "repo.json").exists()
346
347 def test_template_ignored_for_bare_repo(self, tmp_path: pathlib.Path) -> None:
348 tmpl = tmp_path / "tmpl"
349 tmpl.mkdir()
350 (tmpl / "README.md").write_text("hello")
351
352 repo = tmp_path / "repo"
353 repo.mkdir()
354 _init(repo, "--bare", "--template", str(tmpl))
355
356 # --bare suppresses template copy
357 assert not (repo / "README.md").exists()
358
359
360 # ---------------------------------------------------------------------------
361 # JSON output — agent UX
362 # ---------------------------------------------------------------------------
363
364
365 class TestJsonOutput:
366 def test_json_exit_zero(self, tmp_path: pathlib.Path) -> None:
367 result = _init(tmp_path, "--json")
368 assert result.exit_code == 0
369
370 def test_json_is_valid(self, tmp_path: pathlib.Path) -> None:
371 result = _init(tmp_path, "--json")
372 data = json.loads(result.output)
373 assert isinstance(data, dict)
374
375 def test_json_has_required_fields(self, tmp_path: pathlib.Path) -> None:
376 result = _init(tmp_path, "--json")
377 data = json.loads(result.output)
378 for field in ("repo_id", "branch", "domain", "path", "reinitialised", "bare"):
379 assert field in data, f"missing field: {field}"
380
381 def test_json_repo_id_matches_repo_json(self, tmp_path: pathlib.Path) -> None:
382 result = _init(tmp_path, "--json")
383 data = json.loads(result.output)
384 stored = json.loads((tmp_path / ".muse" / "repo.json").read_text())["repo_id"]
385 assert data["repo_id"] == stored
386
387 def test_json_branch_matches_default_branch(self, tmp_path: pathlib.Path) -> None:
388 result = _init(tmp_path, "--default-branch", "dev", "--json")
389 data = json.loads(result.output)
390 assert data["branch"] == "dev"
391
392 def test_json_domain_matches_domain_arg(self, tmp_path: pathlib.Path) -> None:
393 result = _init(tmp_path, "--domain", "code", "--json")
394 data = json.loads(result.output)
395 assert data["domain"] == "code"
396
397 def test_json_reinitialised_false_on_fresh(self, tmp_path: pathlib.Path) -> None:
398 result = _init(tmp_path, "--json")
399 assert json.loads(result.output)["reinitialised"] is False
400
401 def test_json_reinitialised_true_on_force(self, tmp_path: pathlib.Path) -> None:
402 _init(tmp_path)
403 result = _init(tmp_path, "--force", "--json")
404 assert json.loads(result.output)["reinitialised"] is True
405
406 def test_json_force_preserves_repo_id(self, tmp_path: pathlib.Path) -> None:
407 first = json.loads(_init(tmp_path, "--json").output)["repo_id"]
408 second = json.loads(_init(tmp_path, "--force", "--json").output)["repo_id"]
409 assert first == second
410
411 def test_json_bare_flag_reflects_bare_arg(self, tmp_path: pathlib.Path) -> None:
412 data = json.loads(_init(tmp_path, "--bare", "--json").output)
413 assert data["bare"] is True
414
415 def test_json_non_bare_flag(self, tmp_path: pathlib.Path) -> None:
416 data = json.loads(_init(tmp_path, "--json").output)
417 assert data["bare"] is False
418
419 def test_json_path_is_muse_dir(self, tmp_path: pathlib.Path) -> None:
420 data = json.loads(_init(tmp_path, "--json").output)
421 assert data["path"].endswith(".muse")
422 assert pathlib.Path(data["path"]).is_dir()
423
424 def test_json_no_human_text_on_stdout(self, tmp_path: pathlib.Path) -> None:
425 result = _init(tmp_path, "--json")
426 # Must be parseable JSON with no extra prose
427 data = json.loads(result.output.strip())
428 assert isinstance(data, dict)
429
430
431 # ---------------------------------------------------------------------------
432 # Security
433 # ---------------------------------------------------------------------------
434
435
436 class TestSecurity:
437 def test_ansi_in_branch_error_not_on_stdout(self, tmp_path: pathlib.Path) -> None:
438 """Crafted branch names must not inject ANSI into output."""
439 evil = "\x1b[31mevil\x1b[0m"
440 result = _init(tmp_path, "--default-branch", evil)
441 assert "\x1b" not in result.output
442
443 def test_ansi_in_domain_error_not_on_stdout(self, tmp_path: pathlib.Path) -> None:
444 # Domain validator rejects uppercase/special chars including ANSI sequences
445 evil = "EVIL\x1b[31mred\x1b[0m"
446 result = _init(tmp_path, "--domain", evil)
447 assert "\x1b" not in result.output
448
449 def test_control_chars_in_branch_rejected(self, tmp_path: pathlib.Path) -> None:
450 result = _init(tmp_path, "--default-branch", "branch\x00null")
451 assert result.exit_code != 0
452
453 def test_json_errors_are_clean_json(self, tmp_path: pathlib.Path) -> None:
454 """Every error path with --json must emit valid JSON, not a traceback."""
455 cases = [
456 ["--domain", "Bad-Domain!", "--json"],
457 ["--default-branch", "../traversal", "--json"],
458 ["--template", str(tmp_path / "missing"), "--json"],
459 ]
460 for extra in cases:
461 result = _init(tmp_path / "fresh", *extra)
462 data = json.loads(result.output)
463 assert "error" in data, f"missing 'error' key for args: {extra}"
464
465 def test_reinit_without_force_json_has_error(self, tmp_path: pathlib.Path) -> None:
466 _init(tmp_path)
467 result = _init(tmp_path, "--json")
468 assert result.exit_code != 0
469 data = json.loads(result.output)
470 assert "error" in data
471
472 def test_template_symlink_skipped(self, tmp_path: pathlib.Path) -> None:
473 """Symlinks inside a template directory are silently skipped.
474
475 A template with a symlink to ``/etc/passwd`` (or any path outside the
476 template root) must not be followed. The symlink is dropped and the
477 rest of the template is copied normally.
478 """
479 tmpl = tmp_path / "tmpl"
480 tmpl.mkdir()
481 (tmpl / "legit.txt").write_text("ok")
482 (tmpl / "evil_link").symlink_to("/etc/passwd")
483
484 repo = tmp_path / "repo"
485 repo.mkdir()
486 result = _init(repo, "--template", str(tmpl))
487 assert result.exit_code == 0
488 assert (repo / "legit.txt").exists()
489 assert not (repo / "evil_link").exists()
490
491 def test_template_muse_dir_skipped(self, tmp_path: pathlib.Path) -> None:
492 """A ``.muse/`` directory inside a template is never copied.
493
494 Without this guard, a malicious template could overwrite the freshly
495 created VCS state directory with an attacker-controlled repo_id.
496 """
497 tmpl = tmp_path / "tmpl"
498 tmpl.mkdir()
499 evil_muse = tmpl / ".muse"
500 evil_muse.mkdir()
501 (evil_muse / "repo.json").write_text('{"repo_id": "attacker-id"}')
502 (tmpl / "safe.txt").write_text("safe")
503
504 repo = tmp_path / "repo"
505 repo.mkdir()
506 result = _init(repo, "--template", str(tmpl), "--json")
507 assert result.exit_code == 0
508
509 data = json.loads(result.output)
510 # The repo_id must come from init, not from the evil template
511 real_repo_id = (repo / ".muse" / "repo.json")
512 import json as _json
513 stored = _json.loads(real_repo_id.read_text())
514 assert stored["repo_id"] == data["repo_id"]
515 assert stored["repo_id"] != "attacker-id"
516
517 def test_template_symlink_path_rejected(self, tmp_path: pathlib.Path) -> None:
518 """A ``--template`` path that is itself a symlink is rejected."""
519 real_dir = tmp_path / "real"
520 real_dir.mkdir()
521 link = tmp_path / "link"
522 link.symlink_to(real_dir)
523
524 repo = tmp_path / "repo"
525 repo.mkdir()
526 result = _init(repo, "--template", str(link))
527 assert result.exit_code != 0
528
529 def test_tags_dir_created_at_init(self, tmp_path: pathlib.Path) -> None:
530 """init must create ``.muse/tags/`` so muse tag works immediately."""
531 result = _init(tmp_path)
532 assert result.exit_code == 0
533 assert (tmp_path / ".muse" / "tags").is_dir()
534
535 def test_schema_version_is_integer(self, tmp_path: pathlib.Path) -> None:
536 """schema_version in repo.json must be an integer, not a package version string."""
537 _init(tmp_path)
538 data = json.loads((tmp_path / ".muse" / "repo.json").read_text())
539 assert isinstance(data["schema_version"], int), (
540 f"schema_version should be int, got {type(data['schema_version'])}"
541 )
542
543 def test_json_output_includes_schema_version(self, tmp_path: pathlib.Path) -> None:
544 """--json output must include schema_version as an integer."""
545 result = _init(tmp_path, "--json")
546 data = json.loads(result.output)
547 assert "schema_version" in data
548 assert isinstance(data["schema_version"], int)
549
550
551 # ---------------------------------------------------------------------------
552 # Stress
553 # ---------------------------------------------------------------------------
554
555
556 class TestStress:
557 def test_rapid_sequential_inits(self, tmp_path: pathlib.Path) -> None:
558 """50 sequential inits in different directories must all succeed."""
559 for i in range(50):
560 repo = tmp_path / f"repo_{i:03d}"
561 repo.mkdir()
562 result = _init(repo, "--json")
563 assert result.exit_code == 0, f"init failed for repo_{i:03d}"
564 data = json.loads(result.output)
565 assert "repo_id" in data
566
567 def test_large_template_dir(self, tmp_path: pathlib.Path) -> None:
568 """Template with 200 files copies without error."""
569 tmpl = tmp_path / "big_tmpl"
570 tmpl.mkdir()
571 for i in range(200):
572 (tmpl / f"file_{i:03d}.txt").write_text(f"content {i}")
573
574 repo = tmp_path / "repo"
575 repo.mkdir()
576 result = _init(repo, "--template", str(tmpl))
577 assert result.exit_code == 0
578 assert (repo / "file_000.txt").exists()
579 assert (repo / "file_199.txt").exists()
580
581 def test_reinit_cycle_preserves_repo_id(self, tmp_path: pathlib.Path) -> None:
582 """20 successive --force inits must all return the same repo_id."""
583 first = json.loads(_init(tmp_path, "--json").output)["repo_id"]
584 for _ in range(20):
585 rid = json.loads(_init(tmp_path, "--force", "--json").output)["repo_id"]
586 assert rid == first, "repo_id changed across reinit"
587
588 def test_all_domains_produce_valid_ignore(self) -> None:
589 """Every domain in _MUSEIGNORE_DOMAIN_BLOCKS produces valid TOML."""
590 from muse.cli.commands.init import (
591 _MUSEIGNORE_DOMAIN_BLOCKS,
592 _museignore_template,
593 )
594
595 for domain in list(_MUSEIGNORE_DOMAIN_BLOCKS) + ["custom_domain"]:
596 result = _museignore_template(domain)
597 parsed = tomllib.loads(result)
598 assert isinstance(parsed, dict), f"invalid TOML for domain {domain!r}"
599
600
601 # ---------------------------------------------------------------------------
602 # Unit — module constants
603 # ---------------------------------------------------------------------------
604
605
606 class TestConstants:
607 """Structural invariants on module-level constants in init.py."""
608
609 def test_repo_schema_version_is_positive_int(self) -> None:
610 from muse.cli.commands.init import _REPO_SCHEMA_VERSION
611
612 assert isinstance(_REPO_SCHEMA_VERSION, int)
613 assert _REPO_SCHEMA_VERSION >= 1
614
615 def test_default_config_is_valid_toml(self) -> None:
616 from muse.cli.commands.init import _DEFAULT_CONFIG
617
618 parsed = tomllib.loads(_DEFAULT_CONFIG)
619 assert isinstance(parsed, dict)
620
621 def test_default_config_has_user_section(self) -> None:
622 from muse.cli.commands.init import _DEFAULT_CONFIG
623
624 parsed = tomllib.loads(_DEFAULT_CONFIG)
625 assert "user" in parsed
626
627 def test_default_config_has_remotes_section(self) -> None:
628 from muse.cli.commands.init import _DEFAULT_CONFIG
629
630 parsed = tomllib.loads(_DEFAULT_CONFIG)
631 assert "remotes" in parsed
632
633 def test_bare_config_is_valid_toml(self) -> None:
634 from muse.cli.commands.init import _BARE_CONFIG
635
636 parsed = tomllib.loads(_BARE_CONFIG)
637 assert isinstance(parsed, dict)
638
639 def test_bare_config_has_core_bare_true(self) -> None:
640 from muse.cli.commands.init import _BARE_CONFIG
641
642 parsed = tomllib.loads(_BARE_CONFIG)
643 assert parsed.get("core", {}).get("bare") is True
644
645 def test_init_subdirs_is_superset_of_critical_muse_dirs(self) -> None:
646 """Every directory in _CRITICAL_MUSE_DIRS must appear in _INIT_SUBDIRS.
647
648 If this fails, require_repo()'s _verify_muse_dir_integrity() will
649 never see a freshly-init'd critical directory and cannot protect it.
650 """
651 from muse.cli.commands.init import _INIT_SUBDIRS
652 from muse.core.repo import _CRITICAL_MUSE_DIRS
653
654 init_set = set(_INIT_SUBDIRS)
655 for critical in _CRITICAL_MUSE_DIRS:
656 assert critical in init_set, (
657 f".muse/{critical}/ is in _CRITICAL_MUSE_DIRS "
658 f"but missing from _INIT_SUBDIRS — init will not create it"
659 )
660
661 def test_init_subdirs_contains_no_duplicates(self) -> None:
662 from muse.cli.commands.init import _INIT_SUBDIRS
663
664 assert len(_INIT_SUBDIRS) == len(set(_INIT_SUBDIRS))
665
666 def test_init_subdirs_no_absolute_paths(self) -> None:
667 from muse.cli.commands.init import _INIT_SUBDIRS
668
669 for s in _INIT_SUBDIRS:
670 assert not s.startswith("/"), f"{s!r} is absolute — must be relative"
671
672 def test_museignore_global_patterns_is_list(self) -> None:
673 from muse.cli.commands.init import _MUSEIGNORE_GLOBAL
674
675 parsed = tomllib.loads(_MUSEIGNORE_GLOBAL)
676 assert isinstance(parsed["global"]["patterns"], list)
677 assert len(parsed["global"]["patterns"]) > 0
678
679
680 # ---------------------------------------------------------------------------
681 # Unit — _copy_template directly
682 # ---------------------------------------------------------------------------
683
684
685 class TestCopyTemplate:
686 """Direct unit tests for _copy_template — not mediated by the CLI."""
687
688 def test_empty_template_dir_is_a_no_op(self, tmp_path: pathlib.Path) -> None:
689 from muse.cli.commands.init import _copy_template
690
691 src = tmp_path / "src"
692 dst = tmp_path / "dst"
693 src.mkdir()
694 dst.mkdir()
695 _copy_template(src, dst, [])
696 assert list(dst.iterdir()) == []
697
698 def test_files_are_copied(self, tmp_path: pathlib.Path) -> None:
699 from muse.cli.commands.init import _copy_template
700
701 src = tmp_path / "src"
702 dst = tmp_path / "dst"
703 src.mkdir()
704 dst.mkdir()
705 (src / "hello.txt").write_text("hi")
706 _copy_template(src, dst, [])
707 assert (dst / "hello.txt").read_text() == "hi"
708
709 def test_subdirectory_is_copied_recursively(self, tmp_path: pathlib.Path) -> None:
710 from muse.cli.commands.init import _copy_template
711
712 src = tmp_path / "src"
713 dst = tmp_path / "dst"
714 src.mkdir()
715 dst.mkdir()
716 (src / "sub").mkdir()
717 (src / "sub" / "nested.txt").write_text("deep")
718 _copy_template(src, dst, [])
719 assert (dst / "sub" / "nested.txt").read_text() == "deep"
720
721 def test_deeply_nested_structure_copied(self, tmp_path: pathlib.Path) -> None:
722 from muse.cli.commands.init import _copy_template
723
724 src = tmp_path / "src"
725 dst = tmp_path / "dst"
726 src.mkdir()
727 dst.mkdir()
728 deep = src / "a" / "b" / "c"
729 deep.mkdir(parents=True)
730 (deep / "file.txt").write_text("leaf")
731 _copy_template(src, dst, [])
732 assert (dst / "a" / "b" / "c" / "file.txt").read_text() == "leaf"
733
734 def test_symlinks_are_skipped(self, tmp_path: pathlib.Path) -> None:
735 from muse.cli.commands.init import _copy_template
736
737 src = tmp_path / "src"
738 dst = tmp_path / "dst"
739 src.mkdir()
740 dst.mkdir()
741 (src / "legit.txt").write_text("ok")
742 (src / "evil").symlink_to("/etc/passwd")
743 _copy_template(src, dst, [])
744 assert (dst / "legit.txt").exists()
745 assert not (dst / "evil").exists()
746
747 def test_all_symlinks_dir_copies_nothing(self, tmp_path: pathlib.Path) -> None:
748 from muse.cli.commands.init import _copy_template
749
750 src = tmp_path / "src"
751 dst = tmp_path / "dst"
752 src.mkdir()
753 dst.mkdir()
754 for i in range(5):
755 (src / f"link{i}").symlink_to("/etc/passwd")
756 _copy_template(src, dst, [])
757 assert list(dst.iterdir()) == []
758
759 def test_muse_dir_skipped(self, tmp_path: pathlib.Path) -> None:
760 from muse.cli.commands.init import _copy_template
761
762 src = tmp_path / "src"
763 dst = tmp_path / "dst"
764 src.mkdir()
765 dst.mkdir()
766 evil = src / ".muse"
767 evil.mkdir()
768 (evil / "repo.json").write_text('{"repo_id": "attacker"}')
769 (src / "safe.txt").write_text("ok")
770 _copy_template(src, dst, [])
771 assert not (dst / ".muse").exists()
772 assert (dst / "safe.txt").exists()
773
774 def test_muse_file_also_skipped(self, tmp_path: pathlib.Path) -> None:
775 """A file named .muse (not a dir) is also skipped for safety."""
776 from muse.cli.commands.init import _copy_template
777
778 src = tmp_path / "src"
779 dst = tmp_path / "dst"
780 src.mkdir()
781 dst.mkdir()
782 (src / ".muse").write_text("evil")
783 (src / "ok.txt").write_text("ok")
784 _copy_template(src, dst, [])
785 assert not (dst / ".muse").exists()
786 assert (dst / "ok.txt").exists()
787
788 def test_existing_file_overwritten(self, tmp_path: pathlib.Path) -> None:
789 """shutil.copy2 overwrites existing files in the destination."""
790 from muse.cli.commands.init import _copy_template
791
792 src = tmp_path / "src"
793 dst = tmp_path / "dst"
794 src.mkdir()
795 dst.mkdir()
796 (src / "file.txt").write_text("from template")
797 (dst / "file.txt").write_text("original")
798 _copy_template(src, dst, [])
799 assert (dst / "file.txt").read_text() == "from template"
800
801 def test_multiple_symlinks_all_skipped(self, tmp_path: pathlib.Path) -> None:
802 from muse.cli.commands.init import _copy_template
803
804 src = tmp_path / "src"
805 dst = tmp_path / "dst"
806 src.mkdir()
807 dst.mkdir()
808 (src / "real.txt").write_text("real")
809 for name in ("link1", "link2", "link3"):
810 (src / name).symlink_to(str(src / "real.txt"))
811 _copy_template(src, dst, [])
812 assert (dst / "real.txt").exists()
813 for name in ("link1", "link2", "link3"):
814 assert not (dst / name).exists()
815
816 def test_binary_file_copied_correctly(self, tmp_path: pathlib.Path) -> None:
817 from muse.cli.commands.init import _copy_template
818
819 src = tmp_path / "src"
820 dst = tmp_path / "dst"
821 src.mkdir()
822 dst.mkdir()
823 data = bytes(range(256))
824 (src / "binary.bin").write_bytes(data)
825 _copy_template(src, dst, [])
826 assert (dst / "binary.bin").read_bytes() == data
827
828
829 # ---------------------------------------------------------------------------
830 # Integration — HEAD and ref file format
831 # ---------------------------------------------------------------------------
832
833
834 class TestHeadAndRefFile:
835 """Verify the exact format of .muse/HEAD and branch ref files."""
836
837 def test_head_exact_format(self, tmp_path: pathlib.Path) -> None:
838 """HEAD must be exactly 'ref: refs/heads/main\\n'."""
839 _init(tmp_path)
840 head = (tmp_path / ".muse" / "HEAD").read_text()
841 assert head == "ref: refs/heads/main\n"
842
843 def test_head_exact_format_custom_branch(self, tmp_path: pathlib.Path) -> None:
844 _init(tmp_path, "--default-branch", "dev")
845 head = (tmp_path / ".muse" / "HEAD").read_text()
846 assert head == "ref: refs/heads/dev\n"
847
848 def test_head_updated_on_force_with_different_branch(self, tmp_path: pathlib.Path) -> None:
849 """--force with a new --default-branch updates HEAD."""
850 _init(tmp_path, "--default-branch", "main")
851 _init(tmp_path, "--force", "--default-branch", "dev")
852 head = (tmp_path / ".muse" / "HEAD").read_text()
853 assert head == "ref: refs/heads/dev\n"
854
855 def test_branch_ref_file_exists_after_init(self, tmp_path: pathlib.Path) -> None:
856 _init(tmp_path)
857 ref = tmp_path / ".muse" / "refs" / "heads" / "main"
858 assert ref.exists()
859
860 def test_branch_ref_file_is_empty_on_fresh_init(self, tmp_path: pathlib.Path) -> None:
861 """A fresh repo has no commits — branch ref file must be empty."""
862 _init(tmp_path)
863 ref = tmp_path / ".muse" / "refs" / "heads" / "main"
864 assert ref.read_text() == ""
865
866 def test_custom_branch_ref_file_exists(self, tmp_path: pathlib.Path) -> None:
867 _init(tmp_path, "--default-branch", "feat/new")
868 ref = tmp_path / ".muse" / "refs" / "heads" / "feat" / "new"
869 assert ref.exists()
870
871 def test_refs_dir_created(self, tmp_path: pathlib.Path) -> None:
872 """The .muse/refs/ directory itself must exist (not just refs/heads/)."""
873 _init(tmp_path)
874 assert (tmp_path / ".muse" / "refs").is_dir()
875
876 def test_no_directory_is_a_symlink_after_init(self, tmp_path: pathlib.Path) -> None:
877 """Post-init integrity check: every _INIT_SUBDIRS entry must be a real dir."""
878 from muse.cli.commands.init import _INIT_SUBDIRS
879
880 _init(tmp_path)
881 muse = tmp_path / ".muse"
882 for subdir in _INIT_SUBDIRS:
883 candidate = muse / subdir
884 assert candidate.is_dir(), f".muse/{subdir} is not a directory"
885 assert not candidate.is_symlink(), f".muse/{subdir} is a symlink"
886
887
888 # ---------------------------------------------------------------------------
889 # Integration — config file content
890 # ---------------------------------------------------------------------------
891
892
893 class TestConfigFiles:
894 """Verify content and TOML validity of generated config files."""
895
896 def test_config_toml_is_valid_toml(self, tmp_path: pathlib.Path) -> None:
897 _init(tmp_path)
898 parsed = tomllib.loads((tmp_path / ".muse" / "config.toml").read_text())
899 assert isinstance(parsed, dict)
900
901 def test_config_toml_has_user_section(self, tmp_path: pathlib.Path) -> None:
902 _init(tmp_path)
903 parsed = tomllib.loads((tmp_path / ".muse" / "config.toml").read_text())
904 assert "user" in parsed
905
906 def test_config_toml_has_remotes_section(self, tmp_path: pathlib.Path) -> None:
907 _init(tmp_path)
908 parsed = tomllib.loads((tmp_path / ".muse" / "config.toml").read_text())
909 assert "remotes" in parsed
910
911 def test_bare_config_toml_has_core_bare(self, tmp_path: pathlib.Path) -> None:
912 _init(tmp_path, "--bare")
913 parsed = tomllib.loads((tmp_path / ".muse" / "config.toml").read_text())
914 assert parsed.get("core", {}).get("bare") is True
915
916 def test_config_toml_not_overwritten_by_force(self, tmp_path: pathlib.Path) -> None:
917 """--force must not overwrite an existing config.toml."""
918 _init(tmp_path)
919 config_path = tmp_path / ".muse" / "config.toml"
920 config_path.write_text('[custom]\nkey = "value"\n')
921 _init(tmp_path, "--force")
922 parsed = tomllib.loads(config_path.read_text())
923 assert parsed.get("custom", {}).get("key") == "value"
924
925 def test_museignore_has_correct_domain_section(self, tmp_path: pathlib.Path) -> None:
926 """The [domain.<name>] section must match the --domain flag."""
927 _init(tmp_path, "--domain", "midi")
928 parsed = tomllib.loads((tmp_path / ".museignore").read_text())
929 assert "domain" in parsed
930 assert "midi" in parsed["domain"]
931
932 def test_museignore_code_domain_section_present(self, tmp_path: pathlib.Path) -> None:
933 _init(tmp_path, "--domain", "code")
934 parsed = tomllib.loads((tmp_path / ".museignore").read_text())
935 assert "code" in parsed.get("domain", {})
936
937 def test_museignore_does_not_contain_other_domain_sections(
938 self, tmp_path: pathlib.Path
939 ) -> None:
940 """A code-domain repo must not have a [domain.midi] section."""
941 _init(tmp_path, "--domain", "code")
942 text = (tmp_path / ".museignore").read_text()
943 assert "domain.midi" not in text
944
945 def test_repo_json_is_not_zero_bytes(self, tmp_path: pathlib.Path) -> None:
946 """Atomic write guard: repo.json must not be empty after init."""
947 _init(tmp_path)
948 size = (tmp_path / ".muse" / "repo.json").stat().st_size
949 assert size > 0
950
951 def test_repo_json_created_at_is_utc_iso(self, tmp_path: pathlib.Path) -> None:
952 import datetime
953
954 _init(tmp_path)
955 raw = json.loads((tmp_path / ".muse" / "repo.json").read_text())["created_at"]
956 dt = datetime.datetime.fromisoformat(raw)
957 assert dt.tzinfo is not None # must be timezone-aware
958
959 def test_repo_json_domain_matches_arg(self, tmp_path: pathlib.Path) -> None:
960 _init(tmp_path, "--domain", "midi")
961 raw = json.loads((tmp_path / ".muse" / "repo.json").read_text())
962 assert raw["domain"] == "midi"
963
964
965 # ---------------------------------------------------------------------------
966 # Integration — --force edge cases
967 # ---------------------------------------------------------------------------
968
969
970 class TestForceEdgeCases:
971 """Edge cases in the --force reinit path."""
972
973 def test_force_with_corrupt_repo_json_assigns_new_id(
974 self, tmp_path: pathlib.Path
975 ) -> None:
976 """A corrupt repo.json must not crash --force; init assigns a fresh repo_id."""
977 _init(tmp_path)
978 (tmp_path / ".muse" / "repo.json").write_text("{ NOT VALID JSON !!!")
979 result = _init(tmp_path, "--force", "--json")
980 assert result.exit_code == 0
981 data = json.loads(result.output)
982 assert "repo_id" in data
983 # repo_id must be a sha256: content-addressed ID
984 assert data["repo_id"].startswith("sha256:")
985 assert len(data["repo_id"]) == 71
986
987 def test_force_with_empty_repo_json_assigns_new_id(
988 self, tmp_path: pathlib.Path
989 ) -> None:
990 _init(tmp_path)
991 (tmp_path / ".muse" / "repo.json").write_bytes(b"")
992 result = _init(tmp_path, "--force", "--json")
993 assert result.exit_code == 0
994 new_id = json.loads(result.output)["repo_id"]
995 assert new_id.startswith("sha256:")
996 assert len(new_id) == 71
997
998 def test_force_with_repo_json_missing_repo_id_assigns_new_id(
999 self, tmp_path: pathlib.Path
1000 ) -> None:
1001 _init(tmp_path)
1002 (tmp_path / ".muse" / "repo.json").write_text('{"schema_version": 1}')
1003 result = _init(tmp_path, "--force", "--json")
1004 assert result.exit_code == 0
1005 new_id = json.loads(result.output)["repo_id"]
1006 assert new_id.startswith("sha256:")
1007 assert len(new_id) == 71
1008
1009 def test_force_with_non_string_repo_id_assigns_new_id(
1010 self, tmp_path: pathlib.Path
1011 ) -> None:
1012 """repo_id must be a string — if it isn't, force must assign a new one."""
1013 _init(tmp_path)
1014 (tmp_path / ".muse" / "repo.json").write_text('{"repo_id": 42}')
1015 result = _init(tmp_path, "--force", "--json")
1016 assert result.exit_code == 0
1017 rid = json.loads(result.output)["repo_id"]
1018 assert isinstance(rid, str)
1019 assert rid != "42"
1020
1021 def test_force_preserves_custom_museignore(self, tmp_path: pathlib.Path) -> None:
1022 """--force must not overwrite an existing .museignore."""
1023 _init(tmp_path)
1024 custom = '[global]\npatterns = ["custom_file.log"]\n'
1025 (tmp_path / ".museignore").write_text(custom)
1026 _init(tmp_path, "--force")
1027 assert (tmp_path / ".museignore").read_text() == custom
1028
1029 def test_force_preserves_custom_museattributes(self, tmp_path: pathlib.Path) -> None:
1030 _init(tmp_path)
1031 custom = '[meta]\ndomain = "spacetime"\n'
1032 (tmp_path / ".museattributes").write_text(custom)
1033 _init(tmp_path, "--force")
1034 assert (tmp_path / ".museattributes").read_text() == custom
1035
1036 def test_force_reinitialised_flag_in_json(self, tmp_path: pathlib.Path) -> None:
1037 _init(tmp_path)
1038 result = _init(tmp_path, "--force", "--json")
1039 assert json.loads(result.output)["reinitialised"] is True
1040
1041 def test_force_updates_schema_version_in_repo_json(
1042 self, tmp_path: pathlib.Path
1043 ) -> None:
1044 """After --force, repo.json must have the current schema_version."""
1045 from muse.cli.commands.init import _REPO_SCHEMA_VERSION
1046
1047 _init(tmp_path)
1048 _init(tmp_path, "--force")
1049 raw = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1050 assert raw["schema_version"] == _REPO_SCHEMA_VERSION
1051
1052 def test_force_on_partially_missing_muse_dir(self, tmp_path: pathlib.Path) -> None:
1053 """--force on a .muse/ with some subdirs missing re-creates them."""
1054 _init(tmp_path)
1055 import shutil
1056
1057 shutil.rmtree(tmp_path / ".muse" / "objects")
1058 result = _init(tmp_path, "--force")
1059 assert result.exit_code == 0
1060 assert (tmp_path / ".muse" / "objects").is_dir()
1061
1062
1063 # ---------------------------------------------------------------------------
1064 # Integration — two-repo isolation
1065 # ---------------------------------------------------------------------------
1066
1067
1068 class TestRepoIsolation:
1069 """Multiple repos in sibling directories must be completely independent."""
1070
1071 def test_two_repos_have_different_repo_ids(self, tmp_path: pathlib.Path) -> None:
1072 repo_a = tmp_path / "a"
1073 repo_b = tmp_path / "b"
1074 repo_a.mkdir()
1075 repo_b.mkdir()
1076 id_a = json.loads(_init(repo_a, "--json").output)["repo_id"]
1077 id_b = json.loads(_init(repo_b, "--json").output)["repo_id"]
1078 assert id_a != id_b
1079
1080 def test_one_hundred_repos_have_unique_repo_ids(self, tmp_path: pathlib.Path) -> None:
1081 ids: set[str] = set()
1082 for i in range(100):
1083 repo = tmp_path / f"r{i:03d}"
1084 repo.mkdir()
1085 data = json.loads(_init(repo, "--json").output)
1086 ids.add(data["repo_id"])
1087 assert len(ids) == 100, "UUIDs collided across 100 repos"
1088
1089 def test_init_in_child_does_not_affect_parent(self, tmp_path: pathlib.Path) -> None:
1090 """Initialising a subdirectory must not create .muse/ in the parent."""
1091 child = tmp_path / "child"
1092 child.mkdir()
1093 _init(child)
1094 assert not (tmp_path / ".muse").exists()
1095
1096 def test_sibling_repos_independent_after_reinit(self, tmp_path: pathlib.Path) -> None:
1097 repo_a = tmp_path / "a"
1098 repo_b = tmp_path / "b"
1099 repo_a.mkdir()
1100 repo_b.mkdir()
1101 _init(repo_a)
1102 _init(repo_b)
1103 id_a_orig = json.loads((repo_a / ".muse" / "repo.json").read_text())["repo_id"]
1104 _init(repo_b, "--force")
1105 id_a_after = json.loads((repo_a / ".muse" / "repo.json").read_text())["repo_id"]
1106 assert id_a_orig == id_a_after # reinit of b must not touch a
1107
1108
1109 # ---------------------------------------------------------------------------
1110 # End-to-end — muse commands immediately after init
1111 # ---------------------------------------------------------------------------
1112
1113
1114 class TestEndToEnd:
1115 """After muse init, every core command must work without error."""
1116
1117 def test_status_works_on_fresh_repo(self, tmp_path: pathlib.Path) -> None:
1118 import subprocess
1119
1120 _init(tmp_path)
1121 r = subprocess.run(
1122 ["muse", "status", "--json"],
1123 capture_output=True, text=True, cwd=str(tmp_path),
1124 )
1125 assert r.returncode == 0
1126 data = json.loads(r.stdout)
1127 assert data.get("branch") == "main"
1128
1129 def test_log_works_on_fresh_repo(self, tmp_path: pathlib.Path) -> None:
1130 """muse log on an empty repo must exit 0 (no commits is not an error)."""
1131 import subprocess
1132
1133 _init(tmp_path)
1134 r = subprocess.run(
1135 ["muse", "log", "--json"],
1136 capture_output=True, text=True, cwd=str(tmp_path),
1137 )
1138 assert r.returncode == 0
1139
1140 def test_branch_shows_initial_branch(self, tmp_path: pathlib.Path) -> None:
1141 import subprocess
1142
1143 _init(tmp_path)
1144 r = subprocess.run(
1145 ["muse", "branch"],
1146 capture_output=True, text=True, cwd=str(tmp_path),
1147 )
1148 assert r.returncode == 0
1149 assert "main" in r.stdout
1150
1151 def test_branch_shows_custom_initial_branch(self, tmp_path: pathlib.Path) -> None:
1152 import subprocess
1153
1154 _init(tmp_path, "--default-branch", "dev")
1155 r = subprocess.run(
1156 ["muse", "branch"],
1157 capture_output=True, text=True, cwd=str(tmp_path),
1158 )
1159 assert r.returncode == 0
1160 assert "dev" in r.stdout
1161
1162 def test_first_commit_succeeds(self, tmp_path: pathlib.Path) -> None:
1163 """Full workflow: init → add file → commit."""
1164 import subprocess
1165
1166 _init(tmp_path)
1167 (tmp_path / "hello.py").write_text("print('hello')\n")
1168 add = subprocess.run(
1169 ["muse", "code", "add", "."],
1170 capture_output=True, text=True, cwd=str(tmp_path),
1171 )
1172 assert add.returncode == 0, f"muse code add failed: {add.stderr}"
1173 commit = subprocess.run(
1174 ["muse", "commit", "-m", "first commit"],
1175 capture_output=True, text=True, cwd=str(tmp_path),
1176 )
1177 assert commit.returncode == 0, f"muse commit failed: {commit.stderr}"
1178
1179 def test_tag_command_works_after_init(self, tmp_path: pathlib.Path) -> None:
1180 """tags/ is pre-created at init — muse tag must not fail with 'no dir'."""
1181 import subprocess
1182
1183 _init(tmp_path)
1184 # muse tag list should work even on an empty repo
1185 r = subprocess.run(
1186 ["muse", "tag", "list"],
1187 capture_output=True, text=True, cwd=str(tmp_path),
1188 )
1189 # exit 0 expected — no tags is not an error
1190 assert r.returncode == 0
1191
1192 def test_require_repo_succeeds_immediately_after_init(
1193 self, tmp_path: pathlib.Path
1194 ) -> None:
1195 """require_repo() called from Python must not raise after muse init."""
1196 _init(tmp_path)
1197 saved = os.getcwd()
1198 try:
1199 os.chdir(tmp_path)
1200 from muse.core.repo import require_repo
1201
1202 ctx = require_repo()
1203 assert ctx == tmp_path
1204 finally:
1205 os.chdir(saved)
1206
1207 def test_status_on_custom_domain_repo(self, tmp_path: pathlib.Path) -> None:
1208 """muse status works on a repo with a non-default but registered domain."""
1209 import subprocess
1210
1211 _init(tmp_path, "--domain", "scaffold")
1212 r = subprocess.run(
1213 ["muse", "status", "--json"],
1214 capture_output=True, text=True, cwd=str(tmp_path),
1215 )
1216 assert r.returncode == 0
1217
1218 def test_status_correctly_identifies_branch(self, tmp_path: pathlib.Path) -> None:
1219 import subprocess
1220
1221 _init(tmp_path, "--default-branch", "feat/new-world")
1222 r = subprocess.run(
1223 ["muse", "status", "--json"],
1224 capture_output=True, text=True, cwd=str(tmp_path),
1225 )
1226 data = json.loads(r.stdout)
1227 assert data.get("branch") == "feat/new-world"
1228
1229
1230 # ---------------------------------------------------------------------------
1231 # Security — deeper
1232 # ---------------------------------------------------------------------------
1233
1234
1235 class TestSecurityDeep:
1236 """Additional security scenarios beyond the basic TestSecurity class."""
1237
1238 def test_validation_happens_before_filesystem_is_touched(
1239 self, tmp_path: pathlib.Path
1240 ) -> None:
1241 """If the branch name is invalid, no .muse/ directory must be created."""
1242 result = _init(tmp_path, "--default-branch", "bad branch name!")
1243 assert result.exit_code != 0
1244 assert not (tmp_path / ".muse").exists()
1245
1246 def test_domain_validation_before_filesystem_touched(
1247 self, tmp_path: pathlib.Path
1248 ) -> None:
1249 """If the domain is invalid, no .muse/ directory must be created."""
1250 result = _init(tmp_path, "--domain", "BAD_DOMAIN!")
1251 assert result.exit_code != 0
1252 assert not (tmp_path / ".muse").exists()
1253
1254 def test_template_json_error_path_emits_clean_json(
1255 self, tmp_path: pathlib.Path
1256 ) -> None:
1257 """--template with symlink path + --json must emit valid JSON error."""
1258 real = tmp_path / "real"
1259 real.mkdir()
1260 link = tmp_path / "link"
1261 link.symlink_to(real)
1262 result = _init(tmp_path / "repo", "--template", str(link), "--json")
1263 data = json.loads(result.output)
1264 assert "error" in data
1265
1266 def test_template_symlink_inside_subdir_not_followed(
1267 self, tmp_path: pathlib.Path
1268 ) -> None:
1269 """Symlinks nested inside a template's subdirectory are not followed
1270 (the directory containing them is deep-copied by shutil.copytree
1271 which follows symlinks by default — we only guard at the top level).
1272 This test documents the known behaviour so it is explicit."""
1273 tmpl = tmp_path / "tmpl"
1274 (tmpl / "sub").mkdir(parents=True)
1275 # A real file inside a subdirectory
1276 (tmpl / "sub" / "real.txt").write_text("ok")
1277 repo = tmp_path / "repo"
1278 repo.mkdir()
1279 result = _init(repo, "--template", str(tmpl))
1280 assert result.exit_code == 0
1281 # The subdirectory with the real file is copied
1282 assert (repo / "sub" / "real.txt").exists()
1283
1284 def test_all_error_paths_return_nonzero(self, tmp_path: pathlib.Path) -> None:
1285 """Every known error path must return a non-zero exit code."""
1286 cases = [
1287 ["--default-branch", "bad branch"], # bad branch
1288 ["--domain", "Bad!"], # bad domain
1289 ["--template", str(tmp_path / "nope")], # missing template
1290 ]
1291 for args in cases:
1292 result = _init(tmp_path / "fresh", *args)
1293 assert result.exit_code != 0, f"expected failure for args {args}"
1294
1295 def test_very_long_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
1296 """Branch names longer than the allowed max must be rejected."""
1297 long_name = "a" * 300
1298 result = _init(tmp_path, "--default-branch", long_name)
1299 assert result.exit_code != 0
1300
1301 def test_dot_only_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
1302 result = _init(tmp_path, "--default-branch", ".")
1303 assert result.exit_code != 0
1304
1305 def test_dotdot_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
1306 result = _init(tmp_path, "--default-branch", "..")
1307 assert result.exit_code != 0
1308
1309 def test_reinit_twice_without_force_second_fails(self, tmp_path: pathlib.Path) -> None:
1310 """Two consecutive inits without --force: second must always fail."""
1311 assert _init(tmp_path).exit_code == 0
1312 assert _init(tmp_path).exit_code != 0
1313 assert _init(tmp_path).exit_code != 0
1314
1315
1316 # ---------------------------------------------------------------------------
1317 # Stress — deep and concurrent
1318 # ---------------------------------------------------------------------------
1319
1320
1321 class TestStressDeep:
1322 """Large-scale, concurrent, and adversarial stress tests."""
1323
1324 @pytest.mark.slow
1325 def test_concurrent_inits_to_different_dirs(self, tmp_path: pathlib.Path) -> None:
1326 """50 concurrent threads each init a different directory — no crashes,
1327 no UUID collisions, no cross-repo contamination.
1328
1329 Uses subprocess.run (not the _init helper) because os.chdir() is
1330 process-global and not thread-safe. Each subprocess gets its own
1331 working directory via the ``cwd`` argument.
1332 """
1333 import subprocess
1334
1335 results: list[tuple[int, str]] = []
1336 errors: list[str] = []
1337 lock = threading.Lock()
1338
1339 def do_init(i: int) -> None:
1340 repo = tmp_path / f"concurrent_{i:03d}"
1341 repo.mkdir()
1342 r = subprocess.run(
1343 ["muse", "init", "--json"],
1344 capture_output=True, text=True, cwd=str(repo),
1345 )
1346 with lock:
1347 if r.returncode != 0:
1348 errors.append(f"repo_{i}: exit={r.returncode} out={r.stdout[:100]}")
1349 else:
1350 try:
1351 data = json.loads(r.stdout)
1352 results.append((i, data["repo_id"]))
1353 except Exception as exc:
1354 errors.append(f"repo_{i}: parse error {exc}")
1355
1356 threads = [threading.Thread(target=do_init, args=(i,)) for i in range(50)]
1357 for t in threads:
1358 t.start()
1359 for t in threads:
1360 t.join()
1361
1362 assert not errors, f"init errors: {errors}"
1363 assert len(results) == 50
1364
1365 # All repo_ids must be unique
1366 ids = [rid for _, rid in results]
1367 assert len(set(ids)) == len(ids), "UUID collision among concurrent inits"
1368
1369 @pytest.mark.slow
1370 def test_large_deeply_nested_template(self, tmp_path: pathlib.Path) -> None:
1371 """Template with 10 directories × 50 files each (500 total) copies cleanly."""
1372 tmpl = tmp_path / "big"
1373 for d in range(10):
1374 subdir = tmpl / f"dir_{d:02d}"
1375 subdir.mkdir(parents=True)
1376 for f in range(50):
1377 (subdir / f"file_{f:03d}.txt").write_text(f"d={d} f={f}")
1378
1379 repo = tmp_path / "repo"
1380 repo.mkdir()
1381 result = _init(repo, "--template", str(tmpl))
1382 assert result.exit_code == 0
1383 for d in range(10):
1384 assert (repo / f"dir_{d:02d}" / "file_000.txt").exists()
1385 assert (repo / f"dir_{d:02d}" / "file_049.txt").exists()
1386
1387 @pytest.mark.slow
1388 def test_force_reinit_100_times_preserves_repo_id(
1389 self, tmp_path: pathlib.Path
1390 ) -> None:
1391 """100 consecutive --force inits must all return the identical repo_id."""
1392 first = json.loads(_init(tmp_path, "--json").output)["repo_id"]
1393 for i in range(100):
1394 result = _init(tmp_path, "--force", "--json")
1395 assert result.exit_code == 0, f"failed on iteration {i}"
1396 rid = json.loads(result.output)["repo_id"]
1397 assert rid == first, f"repo_id changed on iteration {i}"
1398
1399 @pytest.mark.slow
1400 def test_mixed_template_stress(self, tmp_path: pathlib.Path) -> None:
1401 """Template with a mix of real files, symlinks, .muse dir, subdirs.
1402 Real files must be copied; everything else silently skipped."""
1403 tmpl = tmp_path / "mixed"
1404 tmpl.mkdir()
1405 # Legitimate files
1406 for i in range(100):
1407 (tmpl / f"real_{i:03d}.txt").write_text(f"content {i}")
1408 # Symlinks — should be skipped
1409 for i in range(20):
1410 (tmpl / f"evil_{i:02d}").symlink_to("/etc/passwd")
1411 # .muse dir — should be skipped
1412 (tmpl / ".muse").mkdir()
1413 (tmpl / ".muse" / "repo.json").write_text('{"repo_id": "evil"}')
1414 # Legitimate subdirectory
1415 sub = tmpl / "legit_sub"
1416 sub.mkdir()
1417 (sub / "nested.txt").write_text("nested content")
1418
1419 repo = tmp_path / "repo"
1420 repo.mkdir()
1421 result = _init(repo, "--template", str(tmpl))
1422 assert result.exit_code == 0
1423
1424 # Real files copied
1425 assert (repo / "real_000.txt").read_text() == "content 0"
1426 assert (repo / "real_099.txt").read_text() == "content 99"
1427 assert (repo / "legit_sub" / "nested.txt").read_text() == "nested content"
1428
1429 # Symlinks not copied
1430 for i in range(20):
1431 assert not (repo / f"evil_{i:02d}").exists()
1432
1433 # .muse not overwritten
1434 stored = json.loads((repo / ".muse" / "repo.json").read_text())
1435 assert stored.get("repo_id") != "evil"
1436
1437 def test_all_required_subdirs_created_consistently(
1438 self, tmp_path: pathlib.Path
1439 ) -> None:
1440 """Verify all required subdirs are consistently created across 20 inits."""
1441 from muse.cli.commands.init import _INIT_SUBDIRS
1442
1443 for i in range(20):
1444 repo = tmp_path / f"repo_{i:02d}"
1445 repo.mkdir()
1446 _init(repo)
1447 muse = repo / ".muse"
1448 for subdir in _INIT_SUBDIRS:
1449 assert (muse / subdir).is_dir(), (
1450 f"repo_{i}: .muse/{subdir} missing"
1451 )
1452
1453
1454 # ---------------------------------------------------------------------------
1455 # Directory argument — muse init <dir> (ergonomics parity with git init <dir>)
1456 # ---------------------------------------------------------------------------
1457
1458
1459 def _init_from(cwd: pathlib.Path, *args: str) -> "InvokeResult":
1460 """Invoke ``muse init`` with CWD set to *cwd*, passing *args* verbatim.
1461
1462 Unlike ``_init``, this helper does NOT pre-create the target directory —
1463 the directory-argument feature is expected to create it.
1464 """
1465 from muse.cli.app import main as cli
1466
1467 cwd.mkdir(parents=True, exist_ok=True)
1468 saved = os.getcwd()
1469 try:
1470 os.chdir(cwd)
1471 return runner.invoke(cli, ["init", *args])
1472 finally:
1473 os.chdir(saved)
1474
1475
1476 class TestDirectoryArgument:
1477 """muse init <dir> must create the directory (if needed) and init inside it."""
1478
1479 def test_absolute_path_creates_dir_and_inits(self, tmp_path: pathlib.Path) -> None:
1480 target = tmp_path / "new_repo"
1481 # target does not exist yet — the command must create it
1482 result = _init_from(tmp_path, str(target))
1483 assert result.exit_code == 0, result.output
1484 assert (target / ".muse").is_dir()
1485
1486 def test_relative_path_creates_dir_and_inits(self, tmp_path: pathlib.Path) -> None:
1487 result = _init_from(tmp_path, "sub_repo")
1488 assert result.exit_code == 0, result.output
1489 assert (tmp_path / "sub_repo" / ".muse").is_dir()
1490
1491 def test_dot_is_equivalent_to_no_arg(self, tmp_path: pathlib.Path) -> None:
1492 """muse init . should initialise in CWD, same as muse init."""
1493 result = _init_from(tmp_path, ".")
1494 assert result.exit_code == 0, result.output
1495 assert (tmp_path / ".muse").is_dir()
1496
1497 def test_existing_empty_dir_is_used(self, tmp_path: pathlib.Path) -> None:
1498 target = tmp_path / "existing"
1499 target.mkdir()
1500 result = _init_from(tmp_path, str(target))
1501 assert result.exit_code == 0, result.output
1502 assert (target / ".muse").is_dir()
1503
1504 def test_cwd_is_not_initialised_when_dir_arg_given(self, tmp_path: pathlib.Path) -> None:
1505 """When a directory argument is provided, CWD must NOT get a .muse/."""
1506 target = tmp_path / "target"
1507 result = _init_from(tmp_path, str(target))
1508 assert result.exit_code == 0, result.output
1509 assert not (tmp_path / ".muse").exists()
1510
1511 def test_deeply_nested_nonexistent_path_created(self, tmp_path: pathlib.Path) -> None:
1512 target = tmp_path / "a" / "b" / "c"
1513 result = _init_from(tmp_path, str(target))
1514 assert result.exit_code == 0, result.output
1515 assert (target / ".muse").is_dir()
1516
1517 def test_json_path_reflects_target_dir(self, tmp_path: pathlib.Path) -> None:
1518 target = tmp_path / "my_repo"
1519 result = _init_from(tmp_path, str(target), "--json")
1520 assert result.exit_code == 0, result.output
1521 data = json.loads(result.output)
1522 assert data["path"].endswith(".muse")
1523 assert pathlib.Path(data["path"]).parent.resolve() == target.resolve()
1524
1525 def test_dir_arg_with_json_output(self, tmp_path: pathlib.Path) -> None:
1526 target = tmp_path / "json_repo"
1527 result = _init_from(tmp_path, str(target), "--json")
1528 assert result.exit_code == 0, result.output
1529 data = json.loads(result.output)
1530 assert data["status"] == "ok"
1531 assert "repo_id" in data
1532
1533 def test_dir_arg_with_all_flags(self, tmp_path: pathlib.Path) -> None:
1534 target = tmp_path / "full_flags"
1535 result = _init_from(
1536 tmp_path, str(target),
1537 "--default-branch", "dev",
1538 "--domain", "code",
1539 "--json",
1540 )
1541 assert result.exit_code == 0, result.output
1542 data = json.loads(result.output)
1543 assert data["branch"] == "dev"
1544 assert data["domain"] == "code"
1545
1546 def test_dir_arg_human_output_shows_target_muse_dir(self, tmp_path: pathlib.Path) -> None:
1547 target = tmp_path / "human_repo"
1548 result = _init_from(tmp_path, str(target))
1549 assert result.exit_code == 0, result.output
1550 muse_dir_str = str(target / ".muse")
1551 assert muse_dir_str in result.output or str(target.resolve() / ".muse") in result.output
1552
1553 def test_force_flag_with_dir_arg(self, tmp_path: pathlib.Path) -> None:
1554 target = tmp_path / "force_repo"
1555 _init_from(tmp_path, str(target))
1556 first_id = json.loads(
1557 (target / ".muse" / "repo.json").read_text()
1558 )["repo_id"]
1559 result = _init_from(tmp_path, str(target), "--force")
1560 assert result.exit_code == 0, result.output
1561 second_id = json.loads(
1562 (target / ".muse" / "repo.json").read_text()
1563 )["repo_id"]
1564 assert first_id == second_id
1565
1566 def test_bare_flag_with_dir_arg(self, tmp_path: pathlib.Path) -> None:
1567 target = tmp_path / "bare_repo"
1568 result = _init_from(tmp_path, str(target), "--bare")
1569 assert result.exit_code == 0, result.output
1570 assert (target / ".muse").is_dir()
1571 assert not (target / ".museignore").exists()
1572
1573 def test_status_works_after_dir_arg_init(self, tmp_path: pathlib.Path) -> None:
1574 """Full round-trip: init with dir arg, then muse status inside it."""
1575 import subprocess
1576
1577 target = tmp_path / "rountrip"
1578 result = _init_from(tmp_path, str(target))
1579 assert result.exit_code == 0, result.output
1580 r = subprocess.run(
1581 ["muse", "status", "--json"],
1582 capture_output=True, text=True, cwd=str(target),
1583 )
1584 assert r.returncode == 0, r.stderr
1585 data = json.loads(r.stdout)
1586 assert data["branch"] == "main"
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 141 days ago