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