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