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