test_fetch.py
python
| 1 | """Tests for ``muse fetch``. |
| 2 | |
| 3 | Covers acceptance criteria: |
| 4 | - ``muse fetch`` with no remote configured exits 1 with instructive message. |
| 5 | - ``muse fetch`` POSTs to ``/fetch`` with correct payload structure. |
| 6 | - Remote-tracking refs (``.muse/remotes/origin/<branch>``) are updated. |
| 7 | - Local branches and muse-work/ are NOT modified. |
| 8 | - ``muse fetch --all`` iterates all configured remotes. |
| 9 | - ``muse fetch --prune`` removes stale remote-tracking refs. |
| 10 | - New-branch report lines include "(new branch)" suffix. |
| 11 | - Up-to-date branches are silently skipped (no redundant output). |
| 12 | |
| 13 | All HTTP calls are mocked — no live network required. |
| 14 | """ |
| 15 | from __future__ import annotations |
| 16 | |
| 17 | import json |
| 18 | import pathlib |
| 19 | from unittest.mock import AsyncMock, MagicMock, patch |
| 20 | |
| 21 | import pytest |
| 22 | import typer |
| 23 | |
| 24 | from maestro.muse_cli.commands.fetch import ( |
| 25 | _fetch_async, |
| 26 | _fetch_remote_async, |
| 27 | _format_fetch_line, |
| 28 | _list_local_remote_tracking_branches, |
| 29 | _remove_remote_tracking_ref, |
| 30 | ) |
| 31 | from maestro.muse_cli.config import get_remote_head, set_remote_head |
| 32 | from maestro.muse_cli.errors import ExitCode |
| 33 | from maestro.muse_cli.hub_client import FetchBranchInfo |
| 34 | |
| 35 | |
| 36 | # --------------------------------------------------------------------------- |
| 37 | # Test fixtures / helpers |
| 38 | # --------------------------------------------------------------------------- |
| 39 | |
| 40 | |
| 41 | def _init_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path: |
| 42 | """Create a minimal .muse/ structure for testing.""" |
| 43 | muse_dir = tmp_path / ".muse" |
| 44 | muse_dir.mkdir() |
| 45 | (muse_dir / "repo.json").write_text( |
| 46 | json.dumps({"repo_id": "test-repo-id"}), encoding="utf-8" |
| 47 | ) |
| 48 | (muse_dir / "HEAD").write_text(f"refs/heads/{branch}", encoding="utf-8") |
| 49 | return tmp_path |
| 50 | |
| 51 | |
| 52 | def _write_config_with_token( |
| 53 | root: pathlib.Path, |
| 54 | remote_url: str, |
| 55 | remote_name: str = "origin", |
| 56 | ) -> None: |
| 57 | muse_dir = root / ".muse" |
| 58 | (muse_dir / "config.toml").write_text( |
| 59 | f'[auth]\ntoken = "test-token"\n\n[remotes.{remote_name}]\nurl = "{remote_url}"\n', |
| 60 | encoding="utf-8", |
| 61 | ) |
| 62 | |
| 63 | |
| 64 | def _write_config_multi_remote( |
| 65 | root: pathlib.Path, |
| 66 | remotes: dict[str, str], |
| 67 | ) -> None: |
| 68 | """Write a config.toml with multiple remotes.""" |
| 69 | muse_dir = root / ".muse" |
| 70 | lines = ['[auth]\ntoken = "test-token"\n'] |
| 71 | for name, url in remotes.items(): |
| 72 | lines.append(f'\n[remotes.{name}]\nurl = "{url}"\n') |
| 73 | (muse_dir / "config.toml").write_text("".join(lines), encoding="utf-8") |
| 74 | |
| 75 | |
| 76 | def _make_hub_fetch_response( |
| 77 | branches: list[dict[str, object]] | None = None, |
| 78 | status_code: int = 200, |
| 79 | ) -> MagicMock: |
| 80 | """Return a mock httpx.Response for the fetch endpoint.""" |
| 81 | mock_resp = MagicMock() |
| 82 | mock_resp.status_code = status_code |
| 83 | mock_resp.json.return_value = {"branches": branches or []} |
| 84 | mock_resp.text = "OK" |
| 85 | return mock_resp |
| 86 | |
| 87 | |
| 88 | def _make_branch_info( |
| 89 | branch: str = "main", |
| 90 | head_commit_id: str = "abc1234567890000", |
| 91 | is_new: bool = False, |
| 92 | ) -> dict[str, object]: |
| 93 | return { |
| 94 | "branch": branch, |
| 95 | "head_commit_id": head_commit_id, |
| 96 | "is_new": is_new, |
| 97 | } |
| 98 | |
| 99 | |
| 100 | def _make_mock_hub(response: MagicMock) -> MagicMock: |
| 101 | mock_hub = MagicMock() |
| 102 | mock_hub.__aenter__ = AsyncMock(return_value=mock_hub) |
| 103 | mock_hub.__aexit__ = AsyncMock(return_value=None) |
| 104 | mock_hub.post = AsyncMock(return_value=response) |
| 105 | return mock_hub |
| 106 | |
| 107 | |
| 108 | # --------------------------------------------------------------------------- |
| 109 | # Regression test: fetch updates remote-tracking refs without modifying workdir |
| 110 | # --------------------------------------------------------------------------- |
| 111 | |
| 112 | |
| 113 | @pytest.mark.anyio |
| 114 | async def test_fetch_updates_remote_tracking_refs_without_modifying_workdir( |
| 115 | tmp_path: pathlib.Path, |
| 116 | ) -> None: |
| 117 | """Regression: fetch must update .muse/remotes/origin/main but NOT touch HEAD or refs/heads/.""" |
| 118 | root = _init_repo(tmp_path) |
| 119 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 120 | |
| 121 | head_commit_id = "deadbeef1234567890abcdef01234567" |
| 122 | mock_response = _make_hub_fetch_response( |
| 123 | branches=[_make_branch_info("main", head_commit_id, is_new=True)] |
| 124 | ) |
| 125 | mock_hub = _make_mock_hub(mock_response) |
| 126 | |
| 127 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 128 | await _fetch_remote_async( |
| 129 | root=root, |
| 130 | remote_name="origin", |
| 131 | branches=[], |
| 132 | prune=False, |
| 133 | ) |
| 134 | |
| 135 | # Remote-tracking ref must be updated |
| 136 | stored = get_remote_head("origin", "main", root) |
| 137 | assert stored == head_commit_id |
| 138 | |
| 139 | # Local HEAD must NOT be modified |
| 140 | head_content = (root / ".muse" / "HEAD").read_text(encoding="utf-8").strip() |
| 141 | assert head_content == "refs/heads/main" |
| 142 | |
| 143 | # Local refs/heads/ must NOT be created |
| 144 | local_ref = root / ".muse" / "refs" / "heads" / "main" |
| 145 | assert not local_ref.exists() |
| 146 | |
| 147 | # muse-work/ must NOT be created |
| 148 | assert not (root / "muse-work").exists() |
| 149 | |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # test_fetch_no_remote_exits_1 |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | |
| 156 | def test_fetch_no_remote_exits_1(tmp_path: pathlib.Path) -> None: |
| 157 | """muse fetch exits 1 with instructive message when no remote is configured.""" |
| 158 | import asyncio |
| 159 | |
| 160 | root = _init_repo(tmp_path) |
| 161 | |
| 162 | with pytest.raises(typer.Exit) as exc_info: |
| 163 | asyncio.run( |
| 164 | _fetch_remote_async( |
| 165 | root=root, |
| 166 | remote_name="origin", |
| 167 | branches=[], |
| 168 | prune=False, |
| 169 | ) |
| 170 | ) |
| 171 | |
| 172 | assert exc_info.value.exit_code == int(ExitCode.USER_ERROR) |
| 173 | |
| 174 | |
| 175 | def test_fetch_no_remote_message_is_instructive( |
| 176 | tmp_path: pathlib.Path, |
| 177 | capsys: pytest.CaptureFixture[str], |
| 178 | ) -> None: |
| 179 | """Fetch with no remote prints a message directing user to run muse remote add.""" |
| 180 | import asyncio |
| 181 | |
| 182 | root = _init_repo(tmp_path) |
| 183 | |
| 184 | with pytest.raises(typer.Exit): |
| 185 | asyncio.run( |
| 186 | _fetch_remote_async( |
| 187 | root=root, |
| 188 | remote_name="origin", |
| 189 | branches=[], |
| 190 | prune=False, |
| 191 | ) |
| 192 | ) |
| 193 | |
| 194 | captured = capsys.readouterr() |
| 195 | assert "muse remote add" in captured.out |
| 196 | |
| 197 | |
| 198 | # --------------------------------------------------------------------------- |
| 199 | # test_fetch_posts_to_hub_fetch_endpoint |
| 200 | # --------------------------------------------------------------------------- |
| 201 | |
| 202 | |
| 203 | @pytest.mark.anyio |
| 204 | async def test_fetch_posts_to_hub_fetch_endpoint(tmp_path: pathlib.Path) -> None: |
| 205 | """muse fetch POSTs to /fetch with the branches list.""" |
| 206 | root = _init_repo(tmp_path) |
| 207 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 208 | |
| 209 | captured_paths: list[str] = [] |
| 210 | captured_payloads: list[dict[str, object]] = [] |
| 211 | |
| 212 | async def _fake_post(path: str, **kwargs: object) -> MagicMock: |
| 213 | captured_paths.append(path) |
| 214 | payload = kwargs.get("json", {}) |
| 215 | if isinstance(payload, dict): |
| 216 | captured_payloads.append(payload) |
| 217 | return _make_hub_fetch_response() |
| 218 | |
| 219 | mock_hub = MagicMock() |
| 220 | mock_hub.__aenter__ = AsyncMock(return_value=mock_hub) |
| 221 | mock_hub.__aexit__ = AsyncMock(return_value=None) |
| 222 | mock_hub.post = _fake_post |
| 223 | |
| 224 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 225 | await _fetch_remote_async( |
| 226 | root=root, |
| 227 | remote_name="origin", |
| 228 | branches=["main", "feature/bass"], |
| 229 | prune=False, |
| 230 | ) |
| 231 | |
| 232 | assert len(captured_paths) == 1 |
| 233 | assert captured_paths[0] == "/fetch" |
| 234 | assert len(captured_payloads) == 1 |
| 235 | payload = captured_payloads[0] |
| 236 | assert "branches" in payload |
| 237 | branches_val = payload["branches"] |
| 238 | assert isinstance(branches_val, list) |
| 239 | assert "main" in branches_val |
| 240 | assert "feature/bass" in branches_val |
| 241 | |
| 242 | |
| 243 | # --------------------------------------------------------------------------- |
| 244 | # test_fetch_updates_remote_head_for_each_branch |
| 245 | # --------------------------------------------------------------------------- |
| 246 | |
| 247 | |
| 248 | @pytest.mark.anyio |
| 249 | async def test_fetch_updates_remote_head_for_each_branch(tmp_path: pathlib.Path) -> None: |
| 250 | """Each branch returned by the Hub gets its remote-tracking ref updated.""" |
| 251 | root = _init_repo(tmp_path) |
| 252 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 253 | |
| 254 | main_id = "aabbccddeeff001122334455" * 2 |
| 255 | feature_id = "112233445566778899aabbcc" * 2 |
| 256 | |
| 257 | mock_response = _make_hub_fetch_response( |
| 258 | branches=[ |
| 259 | _make_branch_info("main", main_id, is_new=True), |
| 260 | _make_branch_info("feature/guitar", feature_id, is_new=True), |
| 261 | ] |
| 262 | ) |
| 263 | mock_hub = _make_mock_hub(mock_response) |
| 264 | |
| 265 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 266 | count = await _fetch_remote_async( |
| 267 | root=root, |
| 268 | remote_name="origin", |
| 269 | branches=[], |
| 270 | prune=False, |
| 271 | ) |
| 272 | |
| 273 | assert count == 2 |
| 274 | assert get_remote_head("origin", "main", root) == main_id |
| 275 | assert get_remote_head("origin", "feature/guitar", root) == feature_id |
| 276 | |
| 277 | |
| 278 | # --------------------------------------------------------------------------- |
| 279 | # test_fetch_skips_already_up_to_date_branches |
| 280 | # --------------------------------------------------------------------------- |
| 281 | |
| 282 | |
| 283 | @pytest.mark.anyio |
| 284 | async def test_fetch_skips_already_up_to_date_branches( |
| 285 | tmp_path: pathlib.Path, |
| 286 | capsys: pytest.CaptureFixture[str], |
| 287 | ) -> None: |
| 288 | """Branches whose remote HEAD hasn't moved are silently skipped (count stays 0).""" |
| 289 | root = _init_repo(tmp_path) |
| 290 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 291 | |
| 292 | existing_head = "existing-head-commit-id1234567890ab" |
| 293 | # Pre-write the remote-tracking ref so fetch sees it as already known |
| 294 | set_remote_head("origin", "main", existing_head, root) |
| 295 | |
| 296 | mock_response = _make_hub_fetch_response( |
| 297 | branches=[_make_branch_info("main", existing_head, is_new=False)] |
| 298 | ) |
| 299 | mock_hub = _make_mock_hub(mock_response) |
| 300 | |
| 301 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 302 | count = await _fetch_remote_async( |
| 303 | root=root, |
| 304 | remote_name="origin", |
| 305 | branches=[], |
| 306 | prune=False, |
| 307 | ) |
| 308 | |
| 309 | assert count == 0 |
| 310 | captured = capsys.readouterr() |
| 311 | # No "From origin" line for an up-to-date branch |
| 312 | assert "From origin" not in captured.out |
| 313 | |
| 314 | |
| 315 | # --------------------------------------------------------------------------- |
| 316 | # test_fetch_new_branch_report_includes_new_branch_suffix |
| 317 | # --------------------------------------------------------------------------- |
| 318 | |
| 319 | |
| 320 | @pytest.mark.anyio |
| 321 | async def test_fetch_new_branch_report_includes_new_branch_suffix( |
| 322 | tmp_path: pathlib.Path, |
| 323 | capsys: pytest.CaptureFixture[str], |
| 324 | ) -> None: |
| 325 | """New branches get a '(new branch)' suffix in the fetch report line.""" |
| 326 | root = _init_repo(tmp_path) |
| 327 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 328 | |
| 329 | new_id = "cafebabe123456789abcdef0" * 2 |
| 330 | |
| 331 | mock_response = _make_hub_fetch_response( |
| 332 | branches=[_make_branch_info("feature/new-bass", new_id, is_new=True)] |
| 333 | ) |
| 334 | mock_hub = _make_mock_hub(mock_response) |
| 335 | |
| 336 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 337 | await _fetch_remote_async( |
| 338 | root=root, |
| 339 | remote_name="origin", |
| 340 | branches=[], |
| 341 | prune=False, |
| 342 | ) |
| 343 | |
| 344 | captured = capsys.readouterr() |
| 345 | assert "new branch" in captured.out |
| 346 | assert "feature/new-bass" in captured.out |
| 347 | assert "origin/feature/new-bass" in captured.out |
| 348 | |
| 349 | |
| 350 | # --------------------------------------------------------------------------- |
| 351 | # test_fetch_prune_removes_stale_remote_tracking_refs |
| 352 | # --------------------------------------------------------------------------- |
| 353 | |
| 354 | |
| 355 | @pytest.mark.anyio |
| 356 | async def test_fetch_prune_removes_stale_remote_tracking_refs( |
| 357 | tmp_path: pathlib.Path, |
| 358 | capsys: pytest.CaptureFixture[str], |
| 359 | ) -> None: |
| 360 | """--prune deletes remote-tracking refs for branches no longer on the remote.""" |
| 361 | root = _init_repo(tmp_path) |
| 362 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 363 | |
| 364 | # Pre-write two remote-tracking refs: one still exists remotely, one is stale |
| 365 | set_remote_head("origin", "main", "active-commit-id", root) |
| 366 | set_remote_head("origin", "deleted-branch", "old-commit-id", root) |
| 367 | |
| 368 | # Remote only reports "main""deleted-branch" has been removed on the remote |
| 369 | mock_response = _make_hub_fetch_response( |
| 370 | branches=[_make_branch_info("main", "active-commit-id-v2", is_new=False)] |
| 371 | ) |
| 372 | mock_hub = _make_mock_hub(mock_response) |
| 373 | |
| 374 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 375 | await _fetch_remote_async( |
| 376 | root=root, |
| 377 | remote_name="origin", |
| 378 | branches=[], |
| 379 | prune=True, |
| 380 | ) |
| 381 | |
| 382 | # Stale ref must be gone |
| 383 | assert get_remote_head("origin", "deleted-branch", root) is None |
| 384 | |
| 385 | # Active ref must still be present (and updated) |
| 386 | assert get_remote_head("origin", "main", root) == "active-commit-id-v2" |
| 387 | |
| 388 | # Prune message must appear |
| 389 | captured = capsys.readouterr() |
| 390 | assert "deleted-branch" in captured.out |
| 391 | |
| 392 | |
| 393 | @pytest.mark.anyio |
| 394 | async def test_fetch_prune_noop_when_no_stale_refs(tmp_path: pathlib.Path) -> None: |
| 395 | """--prune is a no-op when all local remote-tracking refs exist on the remote.""" |
| 396 | root = _init_repo(tmp_path) |
| 397 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 398 | |
| 399 | set_remote_head("origin", "main", "some-commit-id", root) |
| 400 | |
| 401 | mock_response = _make_hub_fetch_response( |
| 402 | branches=[_make_branch_info("main", "some-commit-id-v2", is_new=False)] |
| 403 | ) |
| 404 | mock_hub = _make_mock_hub(mock_response) |
| 405 | |
| 406 | with patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub): |
| 407 | await _fetch_remote_async( |
| 408 | root=root, |
| 409 | remote_name="origin", |
| 410 | branches=[], |
| 411 | prune=True, |
| 412 | ) |
| 413 | |
| 414 | # main ref is updated |
| 415 | assert get_remote_head("origin", "main", root) == "some-commit-id-v2" |
| 416 | |
| 417 | |
| 418 | # --------------------------------------------------------------------------- |
| 419 | # test_fetch_all_iterates_all_remotes |
| 420 | # --------------------------------------------------------------------------- |
| 421 | |
| 422 | |
| 423 | @pytest.mark.anyio |
| 424 | async def test_fetch_all_iterates_all_remotes(tmp_path: pathlib.Path) -> None: |
| 425 | """--all causes fetch to contact every configured remote.""" |
| 426 | root = _init_repo(tmp_path) |
| 427 | _write_config_multi_remote( |
| 428 | root, |
| 429 | { |
| 430 | "origin": "https://hub.example.com/musehub/repos/r", |
| 431 | "staging": "https://staging.example.com/musehub/repos/r", |
| 432 | }, |
| 433 | ) |
| 434 | |
| 435 | calls: list[str] = [] |
| 436 | |
| 437 | async def _fetch_remote_spy( |
| 438 | *, |
| 439 | root: pathlib.Path, |
| 440 | remote_name: str, |
| 441 | branches: list[str], |
| 442 | prune: bool, |
| 443 | ) -> int: |
| 444 | calls.append(remote_name) |
| 445 | return 1 |
| 446 | |
| 447 | with patch( |
| 448 | "maestro.muse_cli.commands.fetch._fetch_remote_async", |
| 449 | side_effect=_fetch_remote_spy, |
| 450 | ): |
| 451 | await _fetch_async( |
| 452 | root=root, |
| 453 | remote_name="origin", |
| 454 | fetch_all=True, |
| 455 | prune=False, |
| 456 | branches=[], |
| 457 | ) |
| 458 | |
| 459 | assert sorted(calls) == ["origin", "staging"] |
| 460 | |
| 461 | |
| 462 | @pytest.mark.anyio |
| 463 | async def test_fetch_all_no_remotes_exits_1( |
| 464 | tmp_path: pathlib.Path, |
| 465 | capsys: pytest.CaptureFixture[str], |
| 466 | ) -> None: |
| 467 | """--all with no remotes configured exits 1 with instructive message.""" |
| 468 | root = _init_repo(tmp_path) |
| 469 | |
| 470 | with pytest.raises(typer.Exit) as exc_info: |
| 471 | await _fetch_async( |
| 472 | root=root, |
| 473 | remote_name="origin", |
| 474 | fetch_all=True, |
| 475 | prune=False, |
| 476 | branches=[], |
| 477 | ) |
| 478 | |
| 479 | assert exc_info.value.exit_code == int(ExitCode.USER_ERROR) |
| 480 | captured = capsys.readouterr() |
| 481 | assert "muse remote add" in captured.out |
| 482 | |
| 483 | |
| 484 | # --------------------------------------------------------------------------- |
| 485 | # test_fetch_server_error_exits_3 |
| 486 | # --------------------------------------------------------------------------- |
| 487 | |
| 488 | |
| 489 | @pytest.mark.anyio |
| 490 | async def test_fetch_server_error_exits_3(tmp_path: pathlib.Path) -> None: |
| 491 | """Hub returning non-200 causes fetch to exit with code 3.""" |
| 492 | root = _init_repo(tmp_path) |
| 493 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 494 | |
| 495 | error_response = _make_hub_fetch_response(status_code=500) |
| 496 | mock_hub = _make_mock_hub(error_response) |
| 497 | |
| 498 | with ( |
| 499 | patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub), |
| 500 | pytest.raises(typer.Exit) as exc_info, |
| 501 | ): |
| 502 | await _fetch_remote_async( |
| 503 | root=root, |
| 504 | remote_name="origin", |
| 505 | branches=[], |
| 506 | prune=False, |
| 507 | ) |
| 508 | |
| 509 | assert exc_info.value.exit_code == int(ExitCode.INTERNAL_ERROR) |
| 510 | |
| 511 | |
| 512 | @pytest.mark.anyio |
| 513 | async def test_fetch_network_error_exits_3(tmp_path: pathlib.Path) -> None: |
| 514 | """Network error during fetch causes exit with code 3.""" |
| 515 | import httpx |
| 516 | |
| 517 | root = _init_repo(tmp_path) |
| 518 | _write_config_with_token(root, "https://hub.example.com/musehub/repos/r") |
| 519 | |
| 520 | mock_hub = MagicMock() |
| 521 | mock_hub.__aenter__ = AsyncMock(return_value=mock_hub) |
| 522 | mock_hub.__aexit__ = AsyncMock(return_value=None) |
| 523 | mock_hub.post = AsyncMock(side_effect=httpx.ConnectError("connection refused")) |
| 524 | |
| 525 | with ( |
| 526 | patch("maestro.muse_cli.commands.fetch.MuseHubClient", return_value=mock_hub), |
| 527 | pytest.raises(typer.Exit) as exc_info, |
| 528 | ): |
| 529 | await _fetch_remote_async( |
| 530 | root=root, |
| 531 | remote_name="origin", |
| 532 | branches=[], |
| 533 | prune=False, |
| 534 | ) |
| 535 | |
| 536 | assert exc_info.value.exit_code == int(ExitCode.INTERNAL_ERROR) |
| 537 | |
| 538 | |
| 539 | # --------------------------------------------------------------------------- |
| 540 | # Unit tests: _list_local_remote_tracking_branches |
| 541 | # --------------------------------------------------------------------------- |
| 542 | |
| 543 | |
| 544 | def test_list_local_remote_tracking_branches_empty_when_no_dir( |
| 545 | tmp_path: pathlib.Path, |
| 546 | ) -> None: |
| 547 | """Returns empty list when no remotes directory exists.""" |
| 548 | root = _init_repo(tmp_path) |
| 549 | result = _list_local_remote_tracking_branches("origin", root) |
| 550 | assert result == [] |
| 551 | |
| 552 | |
| 553 | def test_list_local_remote_tracking_branches_returns_branch_names( |
| 554 | tmp_path: pathlib.Path, |
| 555 | ) -> None: |
| 556 | """Returns all branch names from remote-tracking ref files.""" |
| 557 | root = _init_repo(tmp_path) |
| 558 | set_remote_head("origin", "main", "abc", root) |
| 559 | set_remote_head("origin", "feature/groove", "def", root) |
| 560 | result = _list_local_remote_tracking_branches("origin", root) |
| 561 | assert sorted(result) == ["feature/groove", "main"] |
| 562 | |
| 563 | |
| 564 | # --------------------------------------------------------------------------- |
| 565 | # Unit tests: _remove_remote_tracking_ref |
| 566 | # --------------------------------------------------------------------------- |
| 567 | |
| 568 | |
| 569 | def test_remove_remote_tracking_ref_deletes_file(tmp_path: pathlib.Path) -> None: |
| 570 | """Removes the tracking pointer file for a specific branch.""" |
| 571 | root = _init_repo(tmp_path) |
| 572 | set_remote_head("origin", "old-branch", "commit-id", root) |
| 573 | assert get_remote_head("origin", "old-branch", root) == "commit-id" |
| 574 | |
| 575 | _remove_remote_tracking_ref("origin", "old-branch", root) |
| 576 | assert get_remote_head("origin", "old-branch", root) is None |
| 577 | |
| 578 | |
| 579 | def test_remove_remote_tracking_ref_noop_when_missing(tmp_path: pathlib.Path) -> None: |
| 580 | """Removing a non-existent ref is a no-op (does not raise).""" |
| 581 | root = _init_repo(tmp_path) |
| 582 | _remove_remote_tracking_ref("origin", "nonexistent-branch", root) |
| 583 | |
| 584 | |
| 585 | # --------------------------------------------------------------------------- |
| 586 | # Unit tests: _format_fetch_line |
| 587 | # --------------------------------------------------------------------------- |
| 588 | |
| 589 | |
| 590 | def test_format_fetch_line_new_branch() -> None: |
| 591 | """New branches include '(new branch)' suffix.""" |
| 592 | info = FetchBranchInfo( |
| 593 | branch="feature/guitar", |
| 594 | head_commit_id="abc1234567890000", |
| 595 | is_new=True, |
| 596 | ) |
| 597 | line = _format_fetch_line("origin", info) |
| 598 | assert "origin" in line |
| 599 | assert "feature/guitar" in line |
| 600 | assert "origin/feature/guitar" in line |
| 601 | assert "(new branch)" in line |
| 602 | assert "abc12345" in line |
| 603 | |
| 604 | |
| 605 | def test_format_fetch_line_existing_branch() -> None: |
| 606 | """Existing branches do NOT have '(new branch)' suffix.""" |
| 607 | info = FetchBranchInfo( |
| 608 | branch="main", |
| 609 | head_commit_id="def0987654321000", |
| 610 | is_new=False, |
| 611 | ) |
| 612 | line = _format_fetch_line("origin", info) |
| 613 | assert "origin/main" in line |
| 614 | assert "(new branch)" not in line |
| 615 | assert "def09876" in line |