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