gabriel / muse public
test_cmd_bundle.py python
268 lines 9.1 KB
Raw
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
1 """Tests for ``muse bundle`` subcommands.
2
3 Covers: create (default/have prune), unbundle (ref update), verify (clean/corrupt),
4 list-heads, round-trip, stress: 50-commit bundle.
5 """
6
7 from __future__ import annotations
8
9 import datetime
10 import hashlib
11 import json
12 import pathlib
13
14 import msgpack
15 import pytest
16 from tests.cli_test_helper import CliRunner
17
18 cli = None # argparse migration — CliRunner ignores this arg
19 from muse.core.object_store import write_object
20 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
21 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
22 from muse.core._types import Manifest
23
24 runner = CliRunner()
25
26 _REPO_ID = "bundle-test"
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33
34 def _sha(data: bytes) -> str:
35 return hashlib.sha256(data).hexdigest()
36
37
38 def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path:
39 muse = path / ".muse"
40 for d in ("commits", "snapshots", "objects", "refs/heads"):
41 (muse / d).mkdir(parents=True, exist_ok=True)
42 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
43 (muse / "repo.json").write_text(
44 json.dumps({"repo_id": repo_id, "domain": "midi"}), encoding="utf-8"
45 )
46 return path
47
48
49 def _env(repo: pathlib.Path) -> Manifest:
50 return {"MUSE_REPO_ROOT": str(repo)}
51
52
53 _counter = 0
54
55
56 def _make_commit(
57 root: pathlib.Path,
58 parent_id: str | None = None,
59 content: bytes = b"data",
60 branch: str = "main",
61 ) -> str:
62 global _counter
63 _counter += 1
64 c = content + str(_counter).encode()
65 obj_id = _sha(c)
66 write_object(root, obj_id, c)
67 manifest = {f"f_{_counter}.txt": obj_id}
68 snap_id = compute_snapshot_id(manifest)
69 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
70 committed_at = datetime.datetime.now(datetime.timezone.utc)
71 parent_ids = [parent_id] if parent_id else []
72 commit_id = compute_commit_id(parent_ids, snap_id, f"commit {_counter}", committed_at.isoformat())
73 write_commit(root, CommitRecord(
74 commit_id=commit_id,
75 repo_id=_REPO_ID,
76 branch=branch,
77 snapshot_id=snap_id,
78 message=f"commit {_counter}",
79 committed_at=committed_at,
80 parent_commit_id=parent_id,
81 ))
82 (root / ".muse" / "refs" / "heads" / branch).write_text(commit_id, encoding="utf-8")
83 return commit_id
84
85
86 # ---------------------------------------------------------------------------
87 # Unit: help
88 # ---------------------------------------------------------------------------
89
90
91 def test_bundle_help() -> None:
92 result = runner.invoke(cli, ["bundle", "--help"])
93 assert result.exit_code == 0
94
95
96 def test_bundle_create_help() -> None:
97 result = runner.invoke(cli, ["bundle", "create", "--help"])
98 assert result.exit_code == 0
99
100
101 # ---------------------------------------------------------------------------
102 # Unit: create
103 # ---------------------------------------------------------------------------
104
105
106 def test_bundle_create_basic(tmp_path: pathlib.Path) -> None:
107 _init_repo(tmp_path)
108 _make_commit(tmp_path, content=b"first")
109 out = tmp_path / "out.bundle"
110 result = runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
111 assert result.exit_code == 0
112 assert out.exists()
113 data = msgpack.unpackb(out.read_bytes(), raw=False)
114 assert "commits" in data
115 assert len(data["commits"]) >= 1
116
117
118 def test_bundle_create_no_commits(tmp_path: pathlib.Path) -> None:
119 _init_repo(tmp_path)
120 out = tmp_path / "empty.bundle"
121 result = runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
122 assert result.exit_code != 0 # no commits to bundle
123
124
125 # ---------------------------------------------------------------------------
126 # Unit: verify clean
127 # ---------------------------------------------------------------------------
128
129
130 def test_bundle_verify_clean(tmp_path: pathlib.Path) -> None:
131 _init_repo(tmp_path)
132 _make_commit(tmp_path, content=b"verify me")
133 out = tmp_path / "clean.bundle"
134 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
135 result = runner.invoke(cli, ["bundle", "verify", str(out)], env=_env(tmp_path))
136 assert result.exit_code == 0
137 assert "clean" in result.output.lower()
138
139
140 def test_bundle_verify_corrupt(tmp_path: pathlib.Path) -> None:
141 _init_repo(tmp_path)
142 _make_commit(tmp_path, content=b"to corrupt")
143 out = tmp_path / "corrupt.bundle"
144 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
145
146 # Tamper with an object's content bytes.
147 raw = msgpack.unpackb(out.read_bytes(), raw=False)
148 if raw.get("objects"):
149 raw["objects"][0]["content"] = b"tampered!"
150 out.write_bytes(msgpack.packb(raw, use_bin_type=True))
151
152 result = runner.invoke(cli, ["bundle", "verify", str(out)], env=_env(tmp_path))
153 assert result.exit_code != 0
154 assert "mismatch" in result.output.lower() or "failure" in result.output.lower()
155
156
157 def test_bundle_verify_json(tmp_path: pathlib.Path) -> None:
158 _init_repo(tmp_path)
159 _make_commit(tmp_path, content=b"json verify")
160 out = tmp_path / "jv.bundle"
161 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
162 result = runner.invoke(cli, ["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
163 assert result.exit_code == 0
164 data = json.loads(result.output)
165 assert data["all_ok"] is True
166
167
168 def test_bundle_verify_quiet_clean(tmp_path: pathlib.Path) -> None:
169 _init_repo(tmp_path)
170 _make_commit(tmp_path, content=b"quiet clean")
171 out = tmp_path / "q.bundle"
172 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
173 result = runner.invoke(cli, ["bundle", "verify", str(out), "-q"], env=_env(tmp_path))
174 assert result.exit_code == 0
175
176
177 # ---------------------------------------------------------------------------
178 # Unit: unbundle
179 # ---------------------------------------------------------------------------
180
181
182 def test_bundle_unbundle_writes_objects(tmp_path: pathlib.Path) -> None:
183 src = tmp_path / "src"
184 dst = tmp_path / "dst"
185 src.mkdir()
186 dst.mkdir()
187 _init_repo(src)
188 _init_repo(dst, repo_id="dst-repo")
189 _make_commit(src, content=b"unbundle me")
190
191 out = tmp_path / "unbundle_test.bundle"
192 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(src))
193
194 result = runner.invoke(cli, ["bundle", "unbundle", str(out)], env=_env(dst))
195 assert result.exit_code == 0
196 assert "unpacked" in result.output.lower()
197
198
199 # ---------------------------------------------------------------------------
200 # Unit: list-heads
201 # ---------------------------------------------------------------------------
202
203
204 def test_bundle_list_heads_text(tmp_path: pathlib.Path) -> None:
205 _init_repo(tmp_path)
206 _make_commit(tmp_path, content=b"heads test")
207 out = tmp_path / "heads.bundle"
208 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
209 result = runner.invoke(cli, ["bundle", "list-heads", str(out)], env=_env(tmp_path))
210 assert result.exit_code == 0
211
212
213 def test_bundle_list_heads_json(tmp_path: pathlib.Path) -> None:
214 _init_repo(tmp_path)
215 _make_commit(tmp_path, content=b"json heads")
216 out = tmp_path / "jheads.bundle"
217 runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
218 result = runner.invoke(cli, ["bundle", "list-heads", str(out), "--json"], env=_env(tmp_path))
219 assert result.exit_code == 0
220 json.loads(result.output) # valid JSON
221
222
223 # ---------------------------------------------------------------------------
224 # Integration: full round-trip
225 # ---------------------------------------------------------------------------
226
227
228 def test_bundle_round_trip(tmp_path: pathlib.Path) -> None:
229 """Create a bundle from a source repo, unbundle into a clean target."""
230 src = tmp_path / "src"
231 dst = tmp_path / "dst"
232 src.mkdir()
233 dst.mkdir()
234 _init_repo(src)
235 _init_repo(dst, repo_id="dst-rt")
236
237 prev: str | None = None
238 for i in range(5):
239 prev = _make_commit(src, parent_id=prev, content=f"rt-{i}".encode())
240
241 out = tmp_path / "rt.bundle"
242 create_result = runner.invoke(cli, ["bundle", "create", str(out)], env=_env(src))
243 assert create_result.exit_code == 0
244
245 unbundle_result = runner.invoke(cli, ["bundle", "unbundle", str(out)], env=_env(dst))
246 assert unbundle_result.exit_code == 0
247
248
249 # ---------------------------------------------------------------------------
250 # Stress: 50-commit bundle
251 # ---------------------------------------------------------------------------
252
253
254 def test_bundle_stress_50_commits(tmp_path: pathlib.Path) -> None:
255 _init_repo(tmp_path)
256 prev: str | None = None
257 for i in range(50):
258 prev = _make_commit(tmp_path, parent_id=prev, content=f"stress-{i}".encode())
259
260 out = tmp_path / "stress.bundle"
261 result = runner.invoke(cli, ["bundle", "create", str(out)], env=_env(tmp_path))
262 assert result.exit_code == 0
263
264 raw = msgpack.unpackb(out.read_bytes(), raw=False)
265 assert len(raw.get("commits", [])) == 50
266
267 verify_result = runner.invoke(cli, ["bundle", "verify", str(out), "-q"], env=_env(tmp_path))
268 assert verify_result.exit_code == 0
File History 1 commit
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago