test_branch_supercharge.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
142 days ago
| 1 | """Supercharge tests for ``muse branch``. |
| 2 | |
| 3 | Gaps filled versus the baseline test_cmd_branch.py |
| 4 | --------------------------------------------------- |
| 5 | 1. ``committed_at`` field present in list --json schema (baseline missed it) |
| 6 | 2. ``--sort committeddate`` actual ordering (baseline only checked exit 0) |
| 7 | 3. ``-r`` remote-tracking listing E2E |
| 8 | 4. ``-a`` combined local + remote listing E2E |
| 9 | 5. ``-dr`` remote-tracking ref deletion — success path |
| 10 | 6. ``-vv`` upstream shown in text output |
| 11 | 7. Diamond-merge DAG correctness for ``--merged`` |
| 12 | 8. Data integrity: empty parent dirs cleaned after nested branch delete |
| 13 | 9. Rename into nested path creates parent dirs |
| 14 | 10. Force-rename/copy leaves destination at correct tip |
| 15 | 11. JSON error schemas for delete/rename/copy operations |
| 16 | 12. Performance: ``--sort committeddate`` with 50 branches under 3 s |
| 17 | 13. Security: ANSI injection in ``--merged`` / ``--no-merged`` / ``--contains`` |
| 18 | 14. Docstring coverage for all public helpers |
| 19 | |
| 20 | Test categories |
| 21 | --------------- |
| 22 | - unit : _cleanup_empty_dirs, _ref_file, _list_remotes |
| 23 | - integration : remote ops (-r/-a/-dr), committeddate ordering, DAG, -vv |
| 24 | - e2e : full CLI round-trips for new scenarios |
| 25 | - security : ANSI in filter flags, error output sanitisation |
| 26 | - data_integrity: empty-dir cleanup, atomic rename into deep paths, force overwrites |
| 27 | - performance : --sort committeddate with 50 branches |
| 28 | - docstrings : public helper docstring coverage |
| 29 | """ |
| 30 | |
| 31 | from __future__ import annotations |
| 32 | from collections.abc import Mapping |
| 33 | |
| 34 | import json |
| 35 | import os |
| 36 | import pathlib |
| 37 | import time |
| 38 | |
| 39 | import pytest |
| 40 | |
| 41 | from tests.cli_test_helper import CliRunner, InvokeResult |
| 42 | from muse.core.store import get_head_commit_id, read_current_branch |
| 43 | from muse.core._types import long_id |
| 44 | |
| 45 | runner = CliRunner() |
| 46 | |
| 47 | |
| 48 | # --------------------------------------------------------------------------- |
| 49 | # Helpers |
| 50 | # --------------------------------------------------------------------------- |
| 51 | |
| 52 | |
| 53 | def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult: |
| 54 | saved = os.getcwd() |
| 55 | try: |
| 56 | os.chdir(repo) |
| 57 | return runner.invoke(None, args) |
| 58 | finally: |
| 59 | os.chdir(saved) |
| 60 | |
| 61 | |
| 62 | def _branch(repo: pathlib.Path, *extra: str) -> InvokeResult: |
| 63 | return _invoke(repo, ["branch", *extra]) |
| 64 | |
| 65 | |
| 66 | def _commit(repo: pathlib.Path, *extra: str) -> InvokeResult: |
| 67 | return _invoke(repo, ["commit", *extra]) |
| 68 | |
| 69 | |
| 70 | @pytest.fixture() |
| 71 | def repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 72 | """Initialised repo with one commit on ``main``.""" |
| 73 | saved = os.getcwd() |
| 74 | try: |
| 75 | os.chdir(tmp_path) |
| 76 | runner.invoke(None, ["init"]) |
| 77 | finally: |
| 78 | os.chdir(saved) |
| 79 | (tmp_path / "a.py").write_text("x = 1\n") |
| 80 | _commit(tmp_path, "-m", "initial") |
| 81 | return tmp_path |
| 82 | |
| 83 | |
| 84 | @pytest.fixture() |
| 85 | def two_commit_repo(repo: pathlib.Path) -> pathlib.Path: |
| 86 | (repo / "b.py").write_text("y = 2\n") |
| 87 | _commit(repo, "-m", "second") |
| 88 | return repo |
| 89 | |
| 90 | |
| 91 | def _first_json(result: InvokeResult) -> Mapping[str, object]: |
| 92 | """Extract the first JSON object from mixed stdout+stderr output.""" |
| 93 | for line in result.output.splitlines(): |
| 94 | stripped = line.strip() |
| 95 | if stripped.startswith("{"): |
| 96 | return json.loads(stripped) |
| 97 | raise ValueError(f"No JSON object in output:\n{result.output!r}") |
| 98 | |
| 99 | |
| 100 | def _make_remote_ref( |
| 101 | repo: pathlib.Path, remote: str, branch: str, commit_id: str |
| 102 | ) -> None: |
| 103 | """Write a local remote-tracking ref to simulate a previous push.""" |
| 104 | ref_dir = repo / ".muse" / "remotes" / remote |
| 105 | ref_file = ref_dir / branch |
| 106 | ref_file.parent.mkdir(parents=True, exist_ok=True) |
| 107 | ref_file.write_text(commit_id + "\n", encoding="utf-8") |
| 108 | |
| 109 | |
| 110 | # --------------------------------------------------------------------------- |
| 111 | # Unit: _ref_file |
| 112 | # --------------------------------------------------------------------------- |
| 113 | |
| 114 | |
| 115 | class TestRefFile: |
| 116 | def test_simple_branch(self, tmp_path: pathlib.Path) -> None: |
| 117 | from muse.cli.commands.branch import _ref_file |
| 118 | p = _ref_file(tmp_path, "main") |
| 119 | assert p == tmp_path / ".muse" / "refs" / "heads" / "main" |
| 120 | |
| 121 | def test_nested_branch(self, tmp_path: pathlib.Path) -> None: |
| 122 | from muse.cli.commands.branch import _ref_file |
| 123 | p = _ref_file(tmp_path, "feat/sub/task") |
| 124 | assert p == tmp_path / ".muse" / "refs" / "heads" / "feat" / "sub" / "task" |
| 125 | |
| 126 | def test_returns_pathlib_path(self, tmp_path: pathlib.Path) -> None: |
| 127 | from muse.cli.commands.branch import _ref_file |
| 128 | p = _ref_file(tmp_path, "dev") |
| 129 | assert isinstance(p, pathlib.Path) |
| 130 | |
| 131 | |
| 132 | # --------------------------------------------------------------------------- |
| 133 | # Unit: _cleanup_empty_dirs |
| 134 | # --------------------------------------------------------------------------- |
| 135 | |
| 136 | |
| 137 | class TestCleanupEmptyDirs: |
| 138 | def test_removes_empty_parent_dir(self, tmp_path: pathlib.Path) -> None: |
| 139 | from muse.cli.commands.branch import _cleanup_empty_dirs |
| 140 | heads = tmp_path / "heads" |
| 141 | ref = heads / "feat" / "task" |
| 142 | ref.parent.mkdir(parents=True) |
| 143 | ref.write_text("") |
| 144 | ref.unlink() |
| 145 | _cleanup_empty_dirs(ref, heads) |
| 146 | assert not (heads / "feat").exists() |
| 147 | |
| 148 | def test_stops_at_heads_dir(self, tmp_path: pathlib.Path) -> None: |
| 149 | from muse.cli.commands.branch import _cleanup_empty_dirs |
| 150 | heads = tmp_path / "heads" |
| 151 | heads.mkdir() |
| 152 | ref = heads / "main" |
| 153 | ref.write_text("") |
| 154 | ref.unlink() |
| 155 | _cleanup_empty_dirs(ref, heads) |
| 156 | assert heads.exists() |
| 157 | |
| 158 | def test_leaves_non_empty_parent(self, tmp_path: pathlib.Path) -> None: |
| 159 | from muse.cli.commands.branch import _cleanup_empty_dirs |
| 160 | heads = tmp_path / "heads" |
| 161 | ref_a = heads / "feat" / "a" |
| 162 | ref_b = heads / "feat" / "b" |
| 163 | ref_a.parent.mkdir(parents=True) |
| 164 | ref_a.write_text("") |
| 165 | ref_b.write_text("") |
| 166 | ref_a.unlink() |
| 167 | _cleanup_empty_dirs(ref_a, heads) |
| 168 | # feat/ still has b, so it must not be removed |
| 169 | assert (heads / "feat").exists() |
| 170 | assert (heads / "feat" / "b").exists() |
| 171 | |
| 172 | def test_removes_multiple_levels(self, tmp_path: pathlib.Path) -> None: |
| 173 | from muse.cli.commands.branch import _cleanup_empty_dirs |
| 174 | heads = tmp_path / "heads" |
| 175 | ref = heads / "a" / "b" / "c" |
| 176 | ref.parent.mkdir(parents=True) |
| 177 | ref.write_text("") |
| 178 | ref.unlink() |
| 179 | _cleanup_empty_dirs(ref, heads) |
| 180 | assert not (heads / "a").exists() |
| 181 | |
| 182 | |
| 183 | # --------------------------------------------------------------------------- |
| 184 | # Unit: _list_remotes |
| 185 | # --------------------------------------------------------------------------- |
| 186 | |
| 187 | |
| 188 | class TestListRemotes: |
| 189 | def test_empty_when_no_remotes_dir(self, tmp_path: pathlib.Path) -> None: |
| 190 | from muse.cli.commands.branch import _list_remotes |
| 191 | (tmp_path / ".muse").mkdir() |
| 192 | assert _list_remotes(tmp_path) == [] |
| 193 | |
| 194 | def test_lists_single_remote(self, tmp_path: pathlib.Path) -> None: |
| 195 | from muse.cli.commands.branch import _list_remotes |
| 196 | ref = tmp_path / ".muse" / "remotes" / "origin" / "main" |
| 197 | ref.parent.mkdir(parents=True) |
| 198 | ref.write_text(long_id("a" * 64)) |
| 199 | remotes = _list_remotes(tmp_path) |
| 200 | assert "origin/main" in remotes |
| 201 | |
| 202 | def test_lists_nested_remote_branch(self, tmp_path: pathlib.Path) -> None: |
| 203 | from muse.cli.commands.branch import _list_remotes |
| 204 | ref = tmp_path / ".muse" / "remotes" / "origin" / "feat" / "task" |
| 205 | ref.parent.mkdir(parents=True) |
| 206 | ref.write_text(long_id("b" * 64)) |
| 207 | remotes = _list_remotes(tmp_path) |
| 208 | assert "origin/feat/task" in remotes |
| 209 | |
| 210 | def test_skips_hidden_files(self, tmp_path: pathlib.Path) -> None: |
| 211 | from muse.cli.commands.branch import _list_remotes |
| 212 | ref_dir = tmp_path / ".muse" / "remotes" / "origin" |
| 213 | ref_dir.mkdir(parents=True) |
| 214 | (ref_dir / ".hidden").write_text("ignored") |
| 215 | (ref_dir / "main").write_text(long_id("c" * 64)) |
| 216 | remotes = _list_remotes(tmp_path) |
| 217 | assert all(not r.endswith(".hidden") for r in remotes) |
| 218 | |
| 219 | def test_sorted_output(self, tmp_path: pathlib.Path) -> None: |
| 220 | from muse.cli.commands.branch import _list_remotes |
| 221 | for name in ("z-branch", "a-branch", "m-branch"): |
| 222 | ref = tmp_path / ".muse" / "remotes" / "origin" / name |
| 223 | ref.parent.mkdir(parents=True, exist_ok=True) |
| 224 | ref.write_text(long_id("d" * 64)) |
| 225 | remotes = _list_remotes(tmp_path) |
| 226 | assert remotes == sorted(remotes) |
| 227 | |
| 228 | |
| 229 | # --------------------------------------------------------------------------- |
| 230 | # Integration: JSON schema — committed_at field (gap in baseline) |
| 231 | # --------------------------------------------------------------------------- |
| 232 | |
| 233 | |
| 234 | class TestListJsonCommittedAt: |
| 235 | """committed_at must be present in the listing JSON schema.""" |
| 236 | |
| 237 | def test_committed_at_present_in_schema(self, repo: pathlib.Path) -> None: |
| 238 | result = _branch(repo, "--json") |
| 239 | data = json.loads(result.output) |
| 240 | assert len(data) >= 1 |
| 241 | assert "committed_at" in data[0], ( |
| 242 | "committed_at missing from branch --json output" |
| 243 | ) |
| 244 | |
| 245 | def test_committed_at_is_iso8601(self, repo: pathlib.Path) -> None: |
| 246 | import datetime |
| 247 | result = _branch(repo, "--json") |
| 248 | data = json.loads(result.output) |
| 249 | main = next(b for b in data if b["name"] == "main") |
| 250 | ts = main["committed_at"] |
| 251 | assert ts is not None |
| 252 | # Must parse as ISO 8601 |
| 253 | datetime.datetime.fromisoformat(ts) |
| 254 | |
| 255 | def test_committed_at_null_for_empty_branch(self, repo: pathlib.Path) -> None: |
| 256 | """Branches pointing at no commit (empty branch) get null committed_at.""" |
| 257 | # Force an empty branch by writing an empty ref file directly |
| 258 | ref = repo / ".muse" / "refs" / "heads" / "empty-branch" |
| 259 | ref.write_text("") |
| 260 | result = _branch(repo, "--json") |
| 261 | data = json.loads(result.output) |
| 262 | eb = next((b for b in data if b["name"] == "empty-branch"), None) |
| 263 | assert eb is not None |
| 264 | assert eb["committed_at"] is None |
| 265 | |
| 266 | def test_committed_at_schema_complete(self, repo: pathlib.Path) -> None: |
| 267 | result = _branch(repo, "--json") |
| 268 | data = json.loads(result.output) |
| 269 | required = {"name", "current", "commit_id", "committed_at", |
| 270 | "last_message", "upstream"} |
| 271 | missing = required - set(data[0].keys()) |
| 272 | assert not missing, f"branch --json missing fields: {missing}" |
| 273 | |
| 274 | |
| 275 | # --------------------------------------------------------------------------- |
| 276 | # Integration: --sort committeddate actual ordering |
| 277 | # --------------------------------------------------------------------------- |
| 278 | |
| 279 | |
| 280 | class TestSortCommittedDateOrdering: |
| 281 | """--sort committeddate must emit branches newest-first.""" |
| 282 | |
| 283 | def test_newer_branch_first(self, repo: pathlib.Path) -> None: |
| 284 | # Create a branch at initial commit, then add another commit on main |
| 285 | _branch(repo, "older-branch") |
| 286 | (repo / "c.py").write_text("c=3\n") |
| 287 | _commit(repo, "-m", "newer commit") |
| 288 | _branch(repo, "newer-branch") |
| 289 | |
| 290 | result = _branch(repo, "--sort", "committeddate", "--json") |
| 291 | assert result.exit_code == 0 |
| 292 | data = json.loads(result.output) |
| 293 | names = [b["name"] for b in data] |
| 294 | assert names.index("newer-branch") < names.index("older-branch"), ( |
| 295 | f"newer-branch should sort before older-branch; got order: {names}" |
| 296 | ) |
| 297 | |
| 298 | def test_timestamp_order_consistent_on_ties(self, repo: pathlib.Path) -> None: |
| 299 | """Branches at the same commit produce a stable (name-secondary) order.""" |
| 300 | for name in ("z-same", "a-same", "m-same"): |
| 301 | _branch(repo, name) |
| 302 | result = _branch(repo, "--sort", "committeddate", "--json") |
| 303 | data = json.loads(result.output) |
| 304 | assert result.exit_code == 0 |
| 305 | assert len(data) >= 4 |
| 306 | |
| 307 | def test_empty_branches_sorted_last(self, repo: pathlib.Path) -> None: |
| 308 | """Branches with no commit (committed_at=null) come after dated branches.""" |
| 309 | _branch(repo, "real-commit-branch") |
| 310 | # Write an empty ref manually |
| 311 | (repo / ".muse" / "refs" / "heads" / "no-commit").write_text("") |
| 312 | result = _branch(repo, "--sort", "committeddate", "--json") |
| 313 | data = json.loads(result.output) |
| 314 | names_with_ts = [b["name"] for b in data if b.get("committed_at")] |
| 315 | names_without_ts = [b["name"] for b in data if not b.get("committed_at")] |
| 316 | if names_with_ts and names_without_ts: |
| 317 | last_with_ts = data.index(next(b for b in data if b["name"] == names_with_ts[-1])) |
| 318 | first_without_ts = data.index(next(b for b in data if b["name"] == names_without_ts[0])) |
| 319 | assert last_with_ts < first_without_ts, ( |
| 320 | "Branches with no commit should sort after dated branches" |
| 321 | ) |
| 322 | |
| 323 | |
| 324 | # --------------------------------------------------------------------------- |
| 325 | # Integration: remote-tracking branches (-r / -a / -dr) |
| 326 | # --------------------------------------------------------------------------- |
| 327 | |
| 328 | |
| 329 | class TestRemoteTrackingBranches: |
| 330 | def test_list_r_shows_only_remotes(self, repo: pathlib.Path) -> None: |
| 331 | cid = get_head_commit_id(repo, "main") |
| 332 | _make_remote_ref(repo, "origin", "main", cid) |
| 333 | result = _branch(repo, "-r", "--json") |
| 334 | assert result.exit_code == 0 |
| 335 | data = json.loads(result.output) |
| 336 | names = [b["name"] for b in data] |
| 337 | # Remote entries are prefixed with remotes/ |
| 338 | assert all(n.startswith("remotes/") for n in names), ( |
| 339 | f"-r listing must only contain remote entries; got: {names}" |
| 340 | ) |
| 341 | assert "remotes/origin/main" in names |
| 342 | |
| 343 | def test_list_r_excludes_local_branches(self, repo: pathlib.Path) -> None: |
| 344 | cid = get_head_commit_id(repo, "main") |
| 345 | _make_remote_ref(repo, "origin", "main", cid) |
| 346 | _branch(repo, "local-only") |
| 347 | result = _branch(repo, "-r", "--json") |
| 348 | data = json.loads(result.output) |
| 349 | names = [b["name"] for b in data] |
| 350 | assert "local-only" not in names |
| 351 | |
| 352 | def test_list_a_includes_both(self, repo: pathlib.Path) -> None: |
| 353 | cid = get_head_commit_id(repo, "main") |
| 354 | _make_remote_ref(repo, "origin", "dev", cid) |
| 355 | _branch(repo, "local-feat") |
| 356 | result = _branch(repo, "-a", "--json") |
| 357 | assert result.exit_code == 0 |
| 358 | data = json.loads(result.output) |
| 359 | names = [b["name"] for b in data] |
| 360 | assert "main" in names |
| 361 | assert "local-feat" in names |
| 362 | assert "remotes/origin/dev" in names |
| 363 | |
| 364 | def test_list_r_empty_when_no_remotes(self, repo: pathlib.Path) -> None: |
| 365 | result = _branch(repo, "-r", "--json") |
| 366 | assert result.exit_code == 0 |
| 367 | assert json.loads(result.output) == [] |
| 368 | |
| 369 | def test_list_r_nested_remote_branch(self, repo: pathlib.Path) -> None: |
| 370 | cid = get_head_commit_id(repo, "main") |
| 371 | _make_remote_ref(repo, "origin", "feat/task", cid) |
| 372 | result = _branch(repo, "-r", "--json") |
| 373 | data = json.loads(result.output) |
| 374 | names = [b["name"] for b in data] |
| 375 | assert "remotes/origin/feat/task" in names |
| 376 | |
| 377 | def test_list_a_sorted_by_name(self, repo: pathlib.Path) -> None: |
| 378 | cid = get_head_commit_id(repo, "main") |
| 379 | _make_remote_ref(repo, "origin", "z-remote", cid) |
| 380 | _make_remote_ref(repo, "origin", "a-remote", cid) |
| 381 | _branch(repo, "m-local") |
| 382 | result = _branch(repo, "-a", "--json") |
| 383 | data = json.loads(result.output) |
| 384 | names = [b["name"] for b in data] |
| 385 | # Local branches come first (alphabetically), then remotes/ |
| 386 | local_names = [n for n in names if not n.startswith("remotes/")] |
| 387 | assert local_names == sorted(local_names) |
| 388 | |
| 389 | def test_dr_deletes_remote_tracking_ref(self, repo: pathlib.Path) -> None: |
| 390 | cid = get_head_commit_id(repo, "main") |
| 391 | _make_remote_ref(repo, "origin", "stale", cid) |
| 392 | # Verify it's visible |
| 393 | before = json.loads(_branch(repo, "-r", "--json").output) |
| 394 | assert any(b["name"] == "remotes/origin/stale" for b in before) |
| 395 | # Delete it |
| 396 | result = _branch(repo, "-d", "-r", "origin/stale") |
| 397 | assert result.exit_code == 0 |
| 398 | # Gone now |
| 399 | after = json.loads(_branch(repo, "-r", "--json").output) |
| 400 | assert not any(b["name"] == "remotes/origin/stale" for b in after) |
| 401 | |
| 402 | def test_dr_json_schema(self, repo: pathlib.Path) -> None: |
| 403 | cid = get_head_commit_id(repo, "main") |
| 404 | _make_remote_ref(repo, "origin", "old", cid) |
| 405 | result = _branch(repo, "-d", "-r", "origin/old", "--json") |
| 406 | assert result.exit_code == 0 |
| 407 | data = json.loads(result.output) |
| 408 | assert data["action"] == "deleted_remote_tracking" |
| 409 | assert data["remote"] == "origin" |
| 410 | assert data["branch"] == "old" |
| 411 | |
| 412 | def test_dr_remotes_prefix_accepted(self, repo: pathlib.Path) -> None: |
| 413 | """remotes/origin/branch spelling accepted in addition to origin/branch.""" |
| 414 | cid = get_head_commit_id(repo, "main") |
| 415 | _make_remote_ref(repo, "origin", "with-prefix", cid) |
| 416 | result = _branch(repo, "-d", "-r", "remotes/origin/with-prefix") |
| 417 | assert result.exit_code == 0 |
| 418 | |
| 419 | def test_dr_nonexistent_exits_1(self, repo: pathlib.Path) -> None: |
| 420 | result = _branch(repo, "-d", "-r", "origin/ghost") |
| 421 | assert result.exit_code == 1 |
| 422 | |
| 423 | def test_dr_no_slash_exits_1(self, repo: pathlib.Path) -> None: |
| 424 | result = _branch(repo, "-d", "-r", "justaname") |
| 425 | assert result.exit_code == 1 |
| 426 | |
| 427 | |
| 428 | # --------------------------------------------------------------------------- |
| 429 | # Integration: -vv upstream display |
| 430 | # --------------------------------------------------------------------------- |
| 431 | |
| 432 | |
| 433 | class TestVerboseUpstream: |
| 434 | def _set_upstream(self, repo: pathlib.Path, branch: str, |
| 435 | remote: str, remote_branch: str) -> None: |
| 436 | config_path = repo / ".muse" / "config.toml" |
| 437 | existing = config_path.read_text() if config_path.exists() else "" |
| 438 | existing += ( |
| 439 | f'\n[branch."{branch}"]\n' |
| 440 | f'remote = "{remote}"\n' |
| 441 | f'merge = "refs/heads/{remote_branch}"\n' |
| 442 | ) |
| 443 | config_path.write_text(existing) |
| 444 | |
| 445 | def test_vv_shows_upstream(self, repo: pathlib.Path) -> None: |
| 446 | self._set_upstream(repo, "main", "origin", "main") |
| 447 | result = _branch(repo, "-vv") |
| 448 | assert result.exit_code == 0 |
| 449 | assert "origin/main" in result.output |
| 450 | |
| 451 | def test_vv_upstream_in_brackets(self, repo: pathlib.Path) -> None: |
| 452 | self._set_upstream(repo, "main", "origin", "main") |
| 453 | result = _branch(repo, "-vv") |
| 454 | assert "[origin/main]" in result.output |
| 455 | |
| 456 | def test_v_does_not_show_upstream(self, repo: pathlib.Path) -> None: |
| 457 | self._set_upstream(repo, "main", "origin", "main") |
| 458 | result = _branch(repo, "-v") |
| 459 | # -v shows commit SHA + message but NOT upstream brackets |
| 460 | assert "[origin/main]" not in result.output |
| 461 | |
| 462 | def test_vv_no_upstream_no_brackets(self, repo: pathlib.Path) -> None: |
| 463 | result = _branch(repo, "-vv") |
| 464 | assert "[" not in result.output |
| 465 | |
| 466 | |
| 467 | # --------------------------------------------------------------------------- |
| 468 | # Integration: Diamond-merge DAG correctness for --merged |
| 469 | # --------------------------------------------------------------------------- |
| 470 | |
| 471 | |
| 472 | class TestDiamondMergeDag: |
| 473 | """--merged must handle merge commits with two parents (parent2_commit_id).""" |
| 474 | |
| 475 | def test_merged_branch_included_after_diamond_merge( |
| 476 | self, repo: pathlib.Path |
| 477 | ) -> None: |
| 478 | """ |
| 479 | Build diamond: main ← feat-a, main ← feat-b, then merge feat-a into feat-b. |
| 480 | After merging feat-a into main via feat-b, --merged should include feat-a. |
| 481 | |
| 482 | main ── C1 |
| 483 | \\ |
| 484 | feat-a ── C2 |
| 485 | \\ |
| 486 | main (merged feat-a) ── C3 |
| 487 | """ |
| 488 | # Create and diverge feat-a |
| 489 | _branch(repo, "feat-a") |
| 490 | _invoke(repo, ["checkout", "feat-a"]) |
| 491 | (repo / "fa.py").write_text("fa=1\n") |
| 492 | _commit(repo, "-m", "feat-a commit") |
| 493 | _invoke(repo, ["checkout", "main"]) |
| 494 | # Merge feat-a into main |
| 495 | _invoke(repo, ["merge", "feat-a"]) |
| 496 | # Now --merged on main should include feat-a |
| 497 | result = _branch(repo, "--merged", "--json") |
| 498 | assert result.exit_code == 0 |
| 499 | data = json.loads(result.output) |
| 500 | names = [b["name"] for b in data] |
| 501 | assert "feat-a" in names, f"feat-a should be merged into main; got: {names}" |
| 502 | |
| 503 | def test_unmerged_sibling_excluded_from_diamond( |
| 504 | self, repo: pathlib.Path |
| 505 | ) -> None: |
| 506 | """Two branches from same point; merging one doesn't include the other.""" |
| 507 | _branch(repo, "merged-branch") |
| 508 | _branch(repo, "unmerged-branch") |
| 509 | |
| 510 | _invoke(repo, ["checkout", "merged-branch"]) |
| 511 | (repo / "mb.py").write_text("mb=1\n") |
| 512 | _commit(repo, "-m", "merged commit") |
| 513 | |
| 514 | _invoke(repo, ["checkout", "unmerged-branch"]) |
| 515 | (repo / "ub.py").write_text("ub=1\n") |
| 516 | _commit(repo, "-m", "unmerged commit") |
| 517 | |
| 518 | _invoke(repo, ["checkout", "main"]) |
| 519 | _invoke(repo, ["merge", "merged-branch"]) |
| 520 | |
| 521 | result = _branch(repo, "--merged", "--json") |
| 522 | data = json.loads(result.output) |
| 523 | names = [b["name"] for b in data] |
| 524 | assert "merged-branch" in names |
| 525 | assert "unmerged-branch" not in names |
| 526 | |
| 527 | |
| 528 | # --------------------------------------------------------------------------- |
| 529 | # Data integrity: nested branch cleanup + deep rename |
| 530 | # --------------------------------------------------------------------------- |
| 531 | |
| 532 | |
| 533 | class TestDataIntegrityNested: |
| 534 | def test_delete_nested_cleans_parent_dirs(self, repo: pathlib.Path) -> None: |
| 535 | """Deleting feat/sub/task must remove the now-empty feat/sub/ and feat/ dirs.""" |
| 536 | _branch(repo, "feat/sub/task") |
| 537 | result = _branch(repo, "-D", "feat/sub/task") |
| 538 | assert result.exit_code == 0 |
| 539 | heads = repo / ".muse" / "refs" / "heads" |
| 540 | assert not (heads / "feat").exists(), ( |
| 541 | "feat/ directory should be removed after deleting feat/sub/task" |
| 542 | ) |
| 543 | |
| 544 | def test_delete_nested_keeps_sibling_dir(self, repo: pathlib.Path) -> None: |
| 545 | """Deleting one nested branch must not remove a sibling.""" |
| 546 | _branch(repo, "feat/sub/a") |
| 547 | _branch(repo, "feat/sub/b") |
| 548 | _branch(repo, "-D", "feat/sub/a") |
| 549 | heads = repo / ".muse" / "refs" / "heads" |
| 550 | assert (heads / "feat" / "sub" / "b").exists() |
| 551 | |
| 552 | def test_rename_into_nested_path_creates_dirs(self, repo: pathlib.Path) -> None: |
| 553 | """Renaming a flat branch to a nested path must create intermediate dirs.""" |
| 554 | _branch(repo, "flat-branch") |
| 555 | result = _branch(repo, "-m", "flat-branch", "deep/nested/branch") |
| 556 | assert result.exit_code == 0 |
| 557 | heads = repo / ".muse" / "refs" / "heads" |
| 558 | assert (heads / "deep" / "nested" / "branch").is_file() |
| 559 | |
| 560 | def test_force_rename_preserves_tip(self, repo: pathlib.Path) -> None: |
| 561 | """Force-rename must not lose the commit pointer.""" |
| 562 | cid_before = get_head_commit_id(repo, "main") |
| 563 | _branch(repo, "original") |
| 564 | _branch(repo, "destination") |
| 565 | _branch(repo, "-M", "original", "destination") |
| 566 | cid_after = get_head_commit_id(repo, "destination") |
| 567 | assert cid_after == cid_before |
| 568 | |
| 569 | def test_force_copy_preserves_src_tip(self, repo: pathlib.Path) -> None: |
| 570 | """Force-copy must not modify the source branch.""" |
| 571 | cid_src = get_head_commit_id(repo, "main") |
| 572 | _branch(repo, "src-branch") |
| 573 | (repo / "x.py").write_text("x=99\n") |
| 574 | _commit(repo, "-m", "extra commit") |
| 575 | _branch(repo, "dst-branch") |
| 576 | # Force-copy src-branch (old tip) onto dst-branch |
| 577 | _branch(repo, "-C", "src-branch", "dst-branch") |
| 578 | assert get_head_commit_id(repo, "src-branch") == cid_src |
| 579 | |
| 580 | def test_rename_current_branch_updates_head(self, repo: pathlib.Path) -> None: |
| 581 | """Renaming the currently checked-out branch must update HEAD.""" |
| 582 | _invoke(repo, ["checkout", "-b", "temp-branch"]) |
| 583 | _branch(repo, "-m", "temp-branch", "renamed-branch") |
| 584 | assert read_current_branch(repo) == "renamed-branch" |
| 585 | |
| 586 | |
| 587 | # --------------------------------------------------------------------------- |
| 588 | # Data integrity: JSON error schemas |
| 589 | # --------------------------------------------------------------------------- |
| 590 | |
| 591 | |
| 592 | class TestJsonErrorSchemas: |
| 593 | """Mutation errors must emit structured JSON with error + message keys.""" |
| 594 | |
| 595 | def test_delete_not_found_json_schema(self, repo: pathlib.Path) -> None: |
| 596 | result = _branch(repo, "-d", "ghost", "--json") |
| 597 | assert result.exit_code == 1 |
| 598 | data = _first_json(result) |
| 599 | assert "error" in data |
| 600 | assert "message" in data |
| 601 | |
| 602 | def test_delete_not_merged_json_schema(self, repo: pathlib.Path) -> None: |
| 603 | _branch(repo, "unmerged") |
| 604 | _invoke(repo, ["checkout", "unmerged"]) |
| 605 | (repo / "z.py").write_text("z=1\n") |
| 606 | _commit(repo, "-m", "diverge") |
| 607 | _invoke(repo, ["checkout", "main"]) |
| 608 | result = _branch(repo, "-d", "unmerged", "--json") |
| 609 | assert result.exit_code == 1 |
| 610 | data = _first_json(result) |
| 611 | assert data.get("error") == "not_merged" |
| 612 | assert "hint" in data # must tell user about -D |
| 613 | |
| 614 | def test_delete_current_branch_json_schema(self, repo: pathlib.Path) -> None: |
| 615 | result = _branch(repo, "-d", "main", "--json") |
| 616 | assert result.exit_code == 1 |
| 617 | data = _first_json(result) |
| 618 | assert data.get("error") == "current_branch" |
| 619 | |
| 620 | def test_rename_not_found_json_schema(self, repo: pathlib.Path) -> None: |
| 621 | result = _branch(repo, "-m", "ghost", "new", "--json") |
| 622 | assert result.exit_code == 1 |
| 623 | data = _first_json(result) |
| 624 | assert data.get("error") == "not_found" |
| 625 | |
| 626 | def test_rename_already_exists_json_schema(self, repo: pathlib.Path) -> None: |
| 627 | _branch(repo, "a") |
| 628 | _branch(repo, "b") |
| 629 | result = _branch(repo, "-m", "a", "b", "--json") |
| 630 | assert result.exit_code == 1 |
| 631 | data = _first_json(result) |
| 632 | assert data.get("error") == "already_exists" |
| 633 | assert "hint" in data # must tell user about -M |
| 634 | |
| 635 | def test_copy_not_found_json_schema(self, repo: pathlib.Path) -> None: |
| 636 | result = _branch(repo, "-c", "ghost", "copy", "--json") |
| 637 | assert result.exit_code == 1 |
| 638 | data = _first_json(result) |
| 639 | assert data.get("error") == "not_found" |
| 640 | |
| 641 | def test_create_already_exists_json_schema(self, repo: pathlib.Path) -> None: |
| 642 | result = _branch(repo, "main", "--json") |
| 643 | assert result.exit_code == 1 |
| 644 | data = _first_json(result) |
| 645 | assert data.get("error") == "already_exists" |
| 646 | |
| 647 | |
| 648 | # --------------------------------------------------------------------------- |
| 649 | # Integration: create JSON schema |
| 650 | # --------------------------------------------------------------------------- |
| 651 | |
| 652 | |
| 653 | class TestCreateJsonSchema: |
| 654 | def test_create_json_schema_complete(self, repo: pathlib.Path) -> None: |
| 655 | result = _branch(repo, "new-branch", "--json") |
| 656 | assert result.exit_code == 0 |
| 657 | data = json.loads(result.output) |
| 658 | assert data["action"] == "created" |
| 659 | assert "branch" in data |
| 660 | assert "commit_id" in data |
| 661 | assert "from" in data |
| 662 | |
| 663 | def test_create_commit_id_is_sha256_prefixed(self, repo: pathlib.Path) -> None: |
| 664 | result = _branch(repo, "sha-check", "--json") |
| 665 | data = json.loads(result.output) |
| 666 | cid = data.get("commit_id") |
| 667 | assert cid is not None |
| 668 | assert cid.startswith("sha256:"), f"commit_id should have sha256: prefix; got {cid!r}" |
| 669 | |
| 670 | def test_create_from_is_null_when_no_start_point(self, repo: pathlib.Path) -> None: |
| 671 | result = _branch(repo, "no-sp", "--json") |
| 672 | data = json.loads(result.output) |
| 673 | assert data.get("from") is None |
| 674 | |
| 675 | def test_create_from_set_when_start_point_given(self, repo: pathlib.Path) -> None: |
| 676 | cid = get_head_commit_id(repo, "main") |
| 677 | result = _branch(repo, "from-sp", cid, "--json") |
| 678 | data = json.loads(result.output) |
| 679 | assert data.get("from") == cid |
| 680 | |
| 681 | |
| 682 | # --------------------------------------------------------------------------- |
| 683 | # Security: ANSI in filter flags |
| 684 | # --------------------------------------------------------------------------- |
| 685 | |
| 686 | |
| 687 | class TestSecurityFilterFlags: |
| 688 | def _has_ansi(self, s: str) -> bool: |
| 689 | return "\x1b[" in s |
| 690 | |
| 691 | def test_ansi_in_merged_ref_rejected_or_sanitized(self, repo: pathlib.Path) -> None: |
| 692 | result = _branch(repo, "--merged", "\x1b[31mevil\x1b[0m") |
| 693 | assert not self._has_ansi(result.output) |
| 694 | |
| 695 | def test_ansi_in_no_merged_ref(self, repo: pathlib.Path) -> None: |
| 696 | result = _branch(repo, "--no-merged", "\x1b[31mevil\x1b[0m") |
| 697 | assert not self._has_ansi(result.output) |
| 698 | |
| 699 | def test_ansi_in_contains_ref(self, repo: pathlib.Path) -> None: |
| 700 | result = _branch(repo, "--contains", "\x1b[31mevil\x1b[0m") |
| 701 | assert not self._has_ansi(result.output) |
| 702 | |
| 703 | def test_newline_in_branch_name_rejected(self, repo: pathlib.Path) -> None: |
| 704 | result = _branch(repo, "branch\nevil") |
| 705 | assert result.exit_code == 1 |
| 706 | |
| 707 | def test_ansi_in_delete_json_error_sanitized(self, repo: pathlib.Path) -> None: |
| 708 | result = _branch(repo, "-d", "\x1b[31mevil\x1b[0m", "--json") |
| 709 | assert result.exit_code == 1 |
| 710 | assert not self._has_ansi(result.output) |
| 711 | |
| 712 | |
| 713 | # --------------------------------------------------------------------------- |
| 714 | # Performance: --sort committeddate with 50 branches |
| 715 | # --------------------------------------------------------------------------- |
| 716 | |
| 717 | |
| 718 | class TestSortCommittedDatePerformance: |
| 719 | def test_50_branches_committeddate_under_3s(self, repo: pathlib.Path) -> None: |
| 720 | for i in range(50): |
| 721 | _branch(repo, f"perf/branch-{i:03d}") |
| 722 | |
| 723 | start = time.monotonic() |
| 724 | result = _branch(repo, "--sort", "committeddate", "--json") |
| 725 | elapsed = time.monotonic() - start |
| 726 | |
| 727 | assert result.exit_code == 0 |
| 728 | data = json.loads(result.output) |
| 729 | assert len(data) == 51 # main + 50 |
| 730 | assert elapsed < 3.0, f"--sort committeddate with 51 branches took {elapsed:.2f}s" |
| 731 | |
| 732 | |
| 733 | # --------------------------------------------------------------------------- |
| 734 | # Docstrings |
| 735 | # --------------------------------------------------------------------------- |
| 736 | |
| 737 | |
| 738 | class TestDocstrings: |
| 739 | def _has_doc(self, obj) -> bool: |
| 740 | import inspect |
| 741 | doc = inspect.getdoc(obj) |
| 742 | return bool(doc and len(doc.strip()) > 10) |
| 743 | |
| 744 | def test_module_docstring(self) -> None: |
| 745 | import muse.cli.commands.branch as m |
| 746 | assert self._has_doc(m) |
| 747 | |
| 748 | def test_ref_file_docstring(self) -> None: |
| 749 | from muse.cli.commands.branch import _ref_file |
| 750 | assert self._has_doc(_ref_file) |
| 751 | |
| 752 | def test_list_local_branches_docstring(self) -> None: |
| 753 | from muse.cli.commands.branch import _list_local_branches |
| 754 | assert self._has_doc(_list_local_branches) |
| 755 | |
| 756 | def test_list_remotes_docstring(self) -> None: |
| 757 | from muse.cli.commands.branch import _list_remotes |
| 758 | assert self._has_doc(_list_remotes) |
| 759 | |
| 760 | def test_upstream_for_docstring(self) -> None: |
| 761 | from muse.cli.commands.branch import _upstream_for |
| 762 | assert self._has_doc(_upstream_for) |
| 763 | |
| 764 | def test_commit_ancestors_docstring(self) -> None: |
| 765 | from muse.cli.commands.branch import _commit_ancestors |
| 766 | assert self._has_doc(_commit_ancestors) |
| 767 | |
| 768 | def test_is_merged_docstring(self) -> None: |
| 769 | from muse.cli.commands.branch import _is_merged |
| 770 | assert self._has_doc(_is_merged) |
| 771 | |
| 772 | def test_contains_commit_docstring(self) -> None: |
| 773 | from muse.cli.commands.branch import _contains_commit |
| 774 | assert self._has_doc(_contains_commit) |
| 775 | |
| 776 | def test_cleanup_empty_dirs_docstring(self) -> None: |
| 777 | from muse.cli.commands.branch import _cleanup_empty_dirs |
| 778 | assert self._has_doc(_cleanup_empty_dirs) |
| 779 | |
| 780 | def test_resolve_start_point_docstring(self) -> None: |
| 781 | from muse.cli.commands.branch import _resolve_start_point |
| 782 | assert self._has_doc(_resolve_start_point) |
| 783 | |
| 784 | def test_run_docstring(self) -> None: |
| 785 | from muse.cli.commands.branch import run |
| 786 | assert self._has_doc(run) |
File History
2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
142 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c
docs: docstring sprint for-each-ref→hotspots — idiomatic ru…
Sonnet 4.6
patch
148 days ago