gabriel / muse public
test_cmd_worktree.py python
145 lines 5.2 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 20 days ago
1 """Comprehensive tests for ``muse worktree``.
2
3 Covers:
4 - E2E: add, list, remove worktrees
5 - Integration: worktree directory created and removed
6 - Security: sanitized path output
7 """
8
9 from __future__ import annotations
10
11 import datetime
12 import json
13 import pathlib
14
15 import pytest
16 from tests.cli_test_helper import CliRunner
17 from muse.core.types import fake_id
18 from muse.core.paths import heads_dir, muse_dir
19
20 cli = None # argparse migration — CliRunner ignores this arg
21
22 runner = CliRunner()
23
24
25 # ---------------------------------------------------------------------------
26 # Shared helpers
27 # ---------------------------------------------------------------------------
28
29 def _env(root: pathlib.Path) -> Manifest:
30 return {"MUSE_REPO_ROOT": str(root)}
31
32
33 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
34 dot_muse = muse_dir(tmp_path)
35 dot_muse.mkdir()
36 repo_id = fake_id("repo")
37 (dot_muse / "repo.json").write_text(json.dumps({
38 "repo_id": repo_id,
39 "domain": "midi",
40 "default_branch": "main",
41 "created_at": "2025-01-01T00:00:00+00:00",
42 }), encoding="utf-8")
43 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
44 (dot_muse / "refs" / "heads").mkdir(parents=True)
45 (dot_muse / "snapshots").mkdir()
46 (dot_muse / "commits").mkdir()
47 (dot_muse / "objects").mkdir()
48 return tmp_path, repo_id
49
50
51 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "test") -> str:
52 from muse.core.commits import (
53 CommitRecord,
54 write_commit,
55 )
56 from muse.core.snapshots import (
57 SnapshotRecord,
58 write_snapshot,
59 )
60 from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
61
62 ref_file = heads_dir(root) / "main"
63 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
64 manifest: Manifest = {}
65 snap_id = compute_snapshot_id(manifest)
66 committed_at = datetime.datetime.now(datetime.timezone.utc)
67 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
68 snapshot_id=snap_id, message=message,
69 committed_at_iso=committed_at.isoformat(),
70 )
71 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
72 write_commit(root, CommitRecord(
73 commit_id=commit_id, branch="main",
74 snapshot_id=snap_id, message=message, committed_at=committed_at,
75 parent_commit_id=parent_id,
76 ))
77 ref_file.parent.mkdir(parents=True, exist_ok=True)
78 ref_file.write_text(commit_id, encoding="utf-8")
79 return commit_id
80
81
82 # ---------------------------------------------------------------------------
83 # Tests
84 # ---------------------------------------------------------------------------
85
86 class TestWorktreeCLI:
87 def test_worktree_list_empty(self, tmp_path: pathlib.Path) -> None:
88 root, _ = _init_repo(tmp_path)
89 result = runner.invoke(cli, ["worktree", "list"], env=_env(root), catch_exceptions=False)
90 assert result.exit_code == 0
91
92 def test_worktree_add_creates_directory(self, tmp_path: pathlib.Path) -> None:
93 root, repo_id = _init_repo(tmp_path)
94 _make_commit(root, repo_id)
95 result = runner.invoke(
96 cli, ["worktree", "add", "my-wt", "main"],
97 env=_env(root), catch_exceptions=False
98 )
99 assert result.exit_code == 0
100
101 def test_worktree_list_after_add(self, tmp_path: pathlib.Path) -> None:
102 root, repo_id = _init_repo(tmp_path)
103 _make_commit(root, repo_id)
104 runner.invoke(
105 cli, ["worktree", "add", "my-wt2", "main"],
106 env=_env(root), catch_exceptions=False
107 )
108 result = runner.invoke(cli, ["worktree", "list"], env=_env(root), catch_exceptions=False)
109 assert result.exit_code == 0
110
111 def test_worktree_remove(self, tmp_path: pathlib.Path) -> None:
112 root, repo_id = _init_repo(tmp_path)
113 _make_commit(root, repo_id)
114 runner.invoke(
115 cli, ["worktree", "add", "rm-wt", "main"],
116 env=_env(root), catch_exceptions=False
117 )
118 result = runner.invoke(
119 cli, ["worktree", "remove", "rm-wt"],
120 env=_env(root), catch_exceptions=False
121 )
122 assert result.exit_code == 0
123
124 def test_worktree_list_format_json(self, tmp_path: pathlib.Path) -> None:
125 root, _ = _init_repo(tmp_path)
126 result = runner.invoke(
127 cli, ["worktree", "list", "--json"],
128 env=_env(root), catch_exceptions=False
129 )
130 assert result.exit_code == 0
131 data = json.loads(result.output)
132 envelope = data if isinstance(data, dict) else {}
133 worktrees = envelope.get("worktrees", data)
134 assert isinstance(worktrees, list)
135
136 def test_worktree_output_sanitized(self, tmp_path: pathlib.Path) -> None:
137 root, repo_id = _init_repo(tmp_path)
138 _make_commit(root, repo_id)
139 result = runner.invoke(cli, ["worktree", "list"], env=_env(root), catch_exceptions=False)
140 assert "\x1b" not in result.output
141
142 def test_worktree_remove_nonexistent_fails(self, tmp_path: pathlib.Path) -> None:
143 root, _ = _init_repo(tmp_path)
144 result = runner.invoke(cli, ["worktree", "remove", "nonexistent"], env=_env(root))
145 assert result.exit_code != 0
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 20 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor 28 days ago