gabriel / muse public
test_cmd_snapshot.py python
225 lines 7.6 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests for ``muse snapshot`` subcommands.
2
3 Covers: create (text/json), list (empty/populated/limit), show (text/json/prefix),
4 export (tar.gz/zip), short flags, stress: 50 files.
5 """
6
7 from __future__ import annotations
8
9 import json
10 import pathlib
11 import tarfile
12 import zipfile
13
14 import pytest
15
16 from muse.core._types import Manifest
17 from tests.cli_test_helper import CliRunner
18
19 cli = None # argparse migration — CliRunner ignores this arg
20
21 runner = CliRunner()
22
23
24 # ---------------------------------------------------------------------------
25 # Helpers
26 # ---------------------------------------------------------------------------
27
28
29 def _init_repo(path: pathlib.Path) -> pathlib.Path:
30 muse = path / ".muse"
31 for d in ("commits", "snapshots", "objects", "refs/heads"):
32 (muse / d).mkdir(parents=True, exist_ok=True)
33 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
34 (muse / "repo.json").write_text(
35 json.dumps({"repo_id": "snap-test", "domain": "code"}), encoding="utf-8"
36 )
37 return path
38
39
40 def _env(repo: pathlib.Path) -> Manifest:
41 return {"MUSE_REPO_ROOT": str(repo)}
42
43
44 def _create_files(root: pathlib.Path, count: int = 3) -> list[str]:
45 names: list[str] = []
46 for i in range(count):
47 name = f"file_{i}.txt"
48 (root / name).write_text(f"content {i}", encoding="utf-8")
49 names.append(name)
50 return names
51
52
53 # ---------------------------------------------------------------------------
54 # Unit: snapshot create
55 # ---------------------------------------------------------------------------
56
57
58 def test_snapshot_create_help() -> None:
59 result = runner.invoke(cli, ["snapshot", "--help"])
60 assert result.exit_code == 0
61
62
63 def test_snapshot_create_text(tmp_path: pathlib.Path) -> None:
64 _init_repo(tmp_path)
65 _create_files(tmp_path, 3)
66 result = runner.invoke(cli, ["snapshot", "create"], env=_env(tmp_path))
67 assert result.exit_code == 0
68 assert "file(s)" in result.output
69
70
71 def test_snapshot_create_json(tmp_path: pathlib.Path) -> None:
72 _init_repo(tmp_path)
73 _create_files(tmp_path, 2)
74 result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
75 assert result.exit_code == 0
76 data = json.loads(result.output)
77 assert "snapshot_id" in data
78 assert data["file_count"] >= 1
79
80
81 def test_snapshot_create_with_note(tmp_path: pathlib.Path) -> None:
82 _init_repo(tmp_path)
83 _create_files(tmp_path, 1)
84 result = runner.invoke(cli, ["snapshot", "create", "-m", "WIP note"], env=_env(tmp_path))
85 assert result.exit_code == 0
86 assert "WIP note" in result.output
87
88
89 def test_snapshot_create_short_flags(tmp_path: pathlib.Path) -> None:
90 _init_repo(tmp_path)
91 _create_files(tmp_path, 1)
92 result = runner.invoke(cli, ["snapshot", "create", "-m", "test", "--json"], env=_env(tmp_path))
93 assert result.exit_code == 0
94 data = json.loads(result.output)
95 assert "snapshot_id" in data
96
97
98 # ---------------------------------------------------------------------------
99 # Unit: snapshot list
100 # ---------------------------------------------------------------------------
101
102
103 def test_snapshot_list_empty(tmp_path: pathlib.Path) -> None:
104 _init_repo(tmp_path)
105 result = runner.invoke(cli, ["snapshot", "list"], env=_env(tmp_path))
106 assert result.exit_code == 0
107 assert "no snapshots" in result.output.lower()
108
109
110 def test_snapshot_list_after_create(tmp_path: pathlib.Path) -> None:
111 _init_repo(tmp_path)
112 _create_files(tmp_path, 2)
113 runner.invoke(cli, ["snapshot", "create"], env=_env(tmp_path))
114 result = runner.invoke(cli, ["snapshot", "list"], env=_env(tmp_path))
115 assert result.exit_code == 0
116 lines = [l for l in result.output.strip().split("\n") if l.strip()]
117 assert len(lines) >= 1
118
119
120 def test_snapshot_list_json(tmp_path: pathlib.Path) -> None:
121 _init_repo(tmp_path)
122 _create_files(tmp_path, 2)
123 runner.invoke(cli, ["snapshot", "create"], env=_env(tmp_path))
124 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
125 assert result.exit_code == 0
126 data = json.loads(result.output)
127 assert isinstance(data, list)
128 assert len(data) >= 1
129 assert "snapshot_id" in data[0]
130
131
132 def test_snapshot_list_limit(tmp_path: pathlib.Path) -> None:
133 _init_repo(tmp_path)
134 _create_files(tmp_path, 1)
135 for _ in range(5):
136 runner.invoke(cli, ["snapshot", "create"], env=_env(tmp_path))
137 result = runner.invoke(cli, ["snapshot", "list", "-n", "3", "--json"], env=_env(tmp_path))
138 assert result.exit_code == 0
139 data = json.loads(result.output)
140 assert len(data) <= 3
141
142
143 # ---------------------------------------------------------------------------
144 # Unit: snapshot show
145 # ---------------------------------------------------------------------------
146
147
148 def test_snapshot_show_json(tmp_path: pathlib.Path) -> None:
149 _init_repo(tmp_path)
150 _create_files(tmp_path, 2)
151 create_result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
152 snap_id = json.loads(create_result.output)["snapshot_id"]
153 result = runner.invoke(cli, ["snapshot", "show", snap_id], env=_env(tmp_path))
154 assert result.exit_code == 0
155 data = json.loads(result.output)
156 assert data["snapshot_id"] == snap_id
157 assert "manifest" in data
158
159
160 def test_snapshot_show_prefix(tmp_path: pathlib.Path) -> None:
161 _init_repo(tmp_path)
162 _create_files(tmp_path, 1)
163 create_result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
164 snap_id = json.loads(create_result.output)["snapshot_id"]
165 # Use first 12 chars as prefix.
166 result = runner.invoke(cli, ["snapshot", "show", snap_id[:12]], env=_env(tmp_path))
167 assert result.exit_code == 0
168
169
170 def test_snapshot_show_not_found(tmp_path: pathlib.Path) -> None:
171 _init_repo(tmp_path)
172 result = runner.invoke(cli, ["snapshot", "show", "nonexistent"], env=_env(tmp_path))
173 assert result.exit_code != 0
174
175
176 # ---------------------------------------------------------------------------
177 # Unit: snapshot export
178 # ---------------------------------------------------------------------------
179
180
181 def test_snapshot_export_tar_gz(tmp_path: pathlib.Path) -> None:
182 _init_repo(tmp_path)
183 _create_files(tmp_path, 2)
184 create_result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
185 snap_id = json.loads(create_result.output)["snapshot_id"]
186 out_file = tmp_path / "snap.tar.gz"
187 result = runner.invoke(
188 cli,
189 ["snapshot", "export", snap_id, "--output", str(out_file)],
190 env=_env(tmp_path),
191 )
192 assert result.exit_code == 0
193 assert out_file.exists()
194 assert tarfile.is_tarfile(str(out_file))
195
196
197 def test_snapshot_export_zip(tmp_path: pathlib.Path) -> None:
198 _init_repo(tmp_path)
199 _create_files(tmp_path, 2)
200 create_result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
201 snap_id = json.loads(create_result.output)["snapshot_id"]
202 out_file = tmp_path / "snap.zip"
203 result = runner.invoke(
204 cli,
205 ["snapshot", "export", snap_id, "--format", "zip", "--output", str(out_file)],
206 env=_env(tmp_path),
207 )
208 assert result.exit_code == 0
209 assert out_file.exists()
210 assert zipfile.is_zipfile(str(out_file))
211
212
213 # ---------------------------------------------------------------------------
214 # Stress: 50 files
215 # ---------------------------------------------------------------------------
216
217
218 def test_snapshot_create_stress_50_files(tmp_path: pathlib.Path) -> None:
219 _init_repo(tmp_path)
220 for i in range(50):
221 (tmp_path / f"stress_{i}.txt").write_bytes(b"x" * 1024)
222 result = runner.invoke(cli, ["snapshot", "create", "--json"], env=_env(tmp_path))
223 assert result.exit_code == 0
224 data = json.loads(result.output)
225 assert data["file_count"] >= 50
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