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