cgcardona / muse public
test_clone.py python
634 lines 21.6 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse clone``.
2
3 Covers acceptance criteria:
4 - ``muse clone <url>`` creates a new directory with .muse/ initialised.
5 - ``muse clone <url> my-project`` creates ./my-project/.
6 - Directory name is derived from the URL when no directory argument is given.
7 - ``--depth N`` is forwarded to the Hub in the clone request.
8 - ``--branch feature/guitar`` is forwarded and applied to HEAD.
9 - ``--single-track drums`` is forwarded in the request payload.
10 - ``--no-checkout`` skips creation of muse-work/.
11 - ``.muse/config.toml`` has origin remote set to source URL.
12 - Commits returned from Hub are stored in local Postgres.
13 - Remote tracking pointer is written after a successful clone.
14 - Cloning into an existing directory exits 1 with an instructive message.
15
16 All HTTP calls are mocked — no live network required.
17 DB calls are mocked for integration tests; unit-level DB tests use the
18 in-memory SQLite fixture from conftest.py.
19 """
20 from __future__ import annotations
21
22 import asyncio
23 import json
24 import pathlib
25 from unittest.mock import AsyncMock, MagicMock, patch
26
27 import pytest
28
29 from maestro.muse_cli.commands.clone import (
30 _clone_async,
31 _derive_directory_name,
32 _init_muse_dir,
33 )
34 from maestro.muse_cli.config import get_remote, get_remote_head
35 from maestro.muse_cli.db import store_pulled_commit
36 from maestro.muse_cli.errors import ExitCode
37
38
39 # ---------------------------------------------------------------------------
40 # Helpers
41 # ---------------------------------------------------------------------------
42
43
44 def _make_clone_response(
45 repo_id: str = "hub-repo-abc123",
46 default_branch: str = "main",
47 remote_head: str | None = "remote-commit-001",
48 commits: list[dict[str, object]] | None = None,
49 objects: list[dict[str, object]] | None = None,
50 ) -> MagicMock:
51 """Return a mock httpx.Response for the clone endpoint."""
52 mock_resp = MagicMock()
53 mock_resp.status_code = 200
54 mock_resp.json.return_value = {
55 "repo_id": repo_id,
56 "default_branch": default_branch,
57 "remote_head": remote_head,
58 "commits": commits or [],
59 "objects": objects or [],
60 }
61 return mock_resp
62
63
64 def _write_token_config(root: pathlib.Path, url: str) -> None:
65 """Write a config.toml with auth token and origin remote."""
66 muse_dir = root / ".muse"
67 muse_dir.mkdir(parents=True, exist_ok=True)
68 (muse_dir / "config.toml").write_text(
69 f'[auth]\ntoken = "test-token"\n\n[remotes.origin]\nurl = "{url}"\n',
70 encoding="utf-8",
71 )
72
73
74 def _mock_hub(mock_response: MagicMock) -> MagicMock:
75 """Build a mock MuseHubClient that returns *mock_response* for POST /clone."""
76 hub = MagicMock()
77 hub.__aenter__ = AsyncMock(return_value=hub)
78 hub.__aexit__ = AsyncMock(return_value=None)
79 hub.post = AsyncMock(return_value=mock_response)
80 return hub
81
82
83 def _run_clone(
84 *,
85 url: str,
86 directory: str | None,
87 depth: int | None = None,
88 branch: str | None = None,
89 single_track: str | None = None,
90 no_checkout: bool = False,
91 mock_response: MagicMock | None = None,
92 ) -> None:
93 """Run _clone_async with all DB/HTTP layers mocked."""
94 response = mock_response or _make_clone_response()
95 mock_hub = _mock_hub(response)
96
97 mock_session_ctx = MagicMock()
98 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
99 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
100
101 with (
102 patch(
103 "maestro.muse_cli.commands.clone.store_pulled_commit",
104 new=AsyncMock(return_value=False),
105 ),
106 patch(
107 "maestro.muse_cli.commands.clone.store_pulled_object",
108 new=AsyncMock(return_value=False),
109 ),
110 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
111 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=mock_hub),
112 ):
113 asyncio.run(
114 _clone_async(
115 url=url,
116 directory=directory,
117 depth=depth,
118 branch=branch,
119 single_track=single_track,
120 no_checkout=no_checkout,
121 )
122 )
123
124
125 # ---------------------------------------------------------------------------
126 # Unit tests: _derive_directory_name
127 # ---------------------------------------------------------------------------
128
129
130 def test_derive_directory_name_simple_path() -> None:
131 """Last URL path component is used as directory name."""
132 assert _derive_directory_name("https://hub.stori.app/repos/my-project") == "my-project"
133
134
135 def test_derive_directory_name_trailing_slash() -> None:
136 """Trailing slashes are stripped before extraction."""
137 assert _derive_directory_name("https://hub.stori.app/repos/my-project/") == "my-project"
138
139
140 def test_derive_directory_name_bare_root() -> None:
141 """Fallback to 'muse-clone' when URL has no useful last segment."""
142 assert _derive_directory_name("https://hub.stori.app/repos") == "muse-clone"
143
144
145 def test_derive_directory_name_with_uuid_segment() -> None:
146 """UUID-style repo paths preserve the identifier."""
147 result = _derive_directory_name("https://hub.stori.app/repos/abc-123-def")
148 assert result == "abc-123-def"
149
150
151 # ---------------------------------------------------------------------------
152 # Unit tests: _init_muse_dir
153 # ---------------------------------------------------------------------------
154
155
156 def test_init_muse_dir_creates_dot_muse(tmp_path: pathlib.Path) -> None:
157 """_init_muse_dir creates the .muse/ directory tree."""
158 _init_muse_dir(tmp_path, "main", "https://hub.stori.app/repos/foo")
159 assert (tmp_path / ".muse").is_dir()
160 assert (tmp_path / ".muse" / "refs" / "heads").is_dir()
161 assert (tmp_path / ".muse" / "repo.json").is_file()
162 assert (tmp_path / ".muse" / "HEAD").is_file()
163 assert (tmp_path / ".muse" / "config.toml").is_file()
164
165
166 def test_init_muse_dir_head_points_at_branch(tmp_path: pathlib.Path) -> None:
167 """HEAD is written as refs/heads/<branch>."""
168 _init_muse_dir(tmp_path, "feature/guitar", "https://hub.example.com/r")
169 head = (tmp_path / ".muse" / "HEAD").read_text(encoding="utf-8").strip()
170 assert head == "refs/heads/feature/guitar"
171
172
173 def test_init_muse_dir_origin_remote_set(tmp_path: pathlib.Path) -> None:
174 """Origin remote URL is written to config.toml."""
175 url = "https://hub.stori.app/repos/my-project"
176 _init_muse_dir(tmp_path, "main", url)
177 stored = get_remote("origin", tmp_path)
178 assert stored == url
179
180
181 def test_init_muse_dir_repo_json_has_schema_version(tmp_path: pathlib.Path) -> None:
182 """repo.json contains schema_version field."""
183 _init_muse_dir(tmp_path, "main", "https://hub.example.com/r")
184 data = json.loads((tmp_path / ".muse" / "repo.json").read_text())
185 assert "schema_version" in data
186 assert "repo_id" in data
187
188
189 # ---------------------------------------------------------------------------
190 # Regression test: test_clone_creates_repo_with_origin_remote
191 # ---------------------------------------------------------------------------
192
193
194 def test_clone_creates_repo_with_origin_remote(tmp_path: pathlib.Path) -> None:
195 """muse clone creates .muse/ and sets origin remote to the source URL.
196
197 Regression — collaborators had no way to create a local
198 copy of a remote Muse Hub repo from scratch.
199 """
200 url = "https://hub.stori.app/repos/my-project"
201 target = tmp_path / "my-project"
202
203 _run_clone(url=url, directory=str(target))
204
205 assert target.is_dir()
206 assert (target / ".muse").is_dir()
207 stored_origin = get_remote("origin", target)
208 assert stored_origin == url
209
210
211 def test_clone_creates_directory_from_url_name(tmp_path: pathlib.Path) -> None:
212 """muse clone without explicit directory uses URL last component as name."""
213 import os
214
215 url = "https://hub.stori.app/repos/producer-beats"
216
217 old_cwd = os.getcwd()
218 try:
219 os.chdir(tmp_path)
220 _run_clone(url=url, directory=None)
221 finally:
222 os.chdir(old_cwd)
223
224 assert (tmp_path / "producer-beats").is_dir()
225 assert (tmp_path / "producer-beats" / ".muse").is_dir()
226
227
228 def test_clone_explicit_directory(tmp_path: pathlib.Path) -> None:
229 """muse clone <url> my-project creates ./my-project/."""
230 url = "https://hub.stori.app/repos/some-repo"
231 target = tmp_path / "my-custom-name"
232
233 _run_clone(url=url, directory=str(target))
234
235 assert target.is_dir()
236 assert (target / ".muse").is_dir()
237
238
239 def test_clone_existing_directory_exits_1(tmp_path: pathlib.Path) -> None:
240 """muse clone into an existing directory exits 1."""
241 import typer
242
243 url = "https://hub.stori.app/repos/foo"
244 existing = tmp_path / "foo"
245 existing.mkdir()
246
247 with pytest.raises(typer.Exit) as exc_info:
248 _run_clone(url=url, directory=str(existing))
249
250 assert exc_info.value.exit_code == int(ExitCode.USER_ERROR)
251
252
253 def test_clone_existing_directory_message(
254 tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
255 ) -> None:
256 """Clone into existing directory prints an instructive error."""
257 import typer
258
259 url = "https://hub.stori.app/repos/foo"
260 existing = tmp_path / "foo"
261 existing.mkdir()
262
263 with pytest.raises(typer.Exit):
264 _run_clone(url=url, directory=str(existing))
265
266 captured = capsys.readouterr()
267 assert "already exists" in captured.out
268
269
270 # ---------------------------------------------------------------------------
271 # Hub interaction tests
272 # ---------------------------------------------------------------------------
273
274
275 @pytest.mark.anyio
276 async def test_clone_calls_hub_post_clone(tmp_path: pathlib.Path) -> None:
277 """muse clone POSTs to /clone with the correct payload fields."""
278 url = "https://hub.stori.app/repos/collab-track"
279 target = tmp_path / "collab-track"
280
281 captured_payloads: list[dict[str, object]] = []
282 mock_response = _make_clone_response()
283
284 hub = MagicMock()
285 hub.__aenter__ = AsyncMock(return_value=hub)
286 hub.__aexit__ = AsyncMock(return_value=None)
287
288 async def fake_post(path: str, **kwargs: object) -> MagicMock:
289 payload = kwargs.get("json", {})
290 if isinstance(payload, dict):
291 captured_payloads.append(payload)
292 return mock_response
293
294 hub.post = fake_post
295
296 mock_session_ctx = MagicMock()
297 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
298 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
299
300 with (
301 patch(
302 "maestro.muse_cli.commands.clone.store_pulled_commit",
303 new=AsyncMock(return_value=False),
304 ),
305 patch(
306 "maestro.muse_cli.commands.clone.store_pulled_object",
307 new=AsyncMock(return_value=False),
308 ),
309 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
310 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=hub),
311 ):
312 await _clone_async(
313 url=url,
314 directory=str(target),
315 depth=5,
316 branch="main",
317 single_track="drums",
318 no_checkout=False,
319 )
320
321 assert len(captured_payloads) == 1
322 payload = captured_payloads[0]
323 assert payload["depth"] == 5
324 assert payload["branch"] == "main"
325 assert payload["single_track"] == "drums"
326
327
328 def test_clone_depth_forwarded_in_request(tmp_path: pathlib.Path) -> None:
329 """--depth N is forwarded to Hub in the clone request payload."""
330 url = "https://hub.stori.app/repos/shallow-repo"
331 target = tmp_path / "shallow-repo"
332
333 captured: list[dict[str, object]] = []
334 mock_response = _make_clone_response()
335
336 hub = MagicMock()
337 hub.__aenter__ = AsyncMock(return_value=hub)
338 hub.__aexit__ = AsyncMock(return_value=None)
339
340 async def fake_post(path: str, **kwargs: object) -> MagicMock:
341 payload = kwargs.get("json", {})
342 if isinstance(payload, dict):
343 captured.append(payload)
344 return mock_response
345
346 hub.post = fake_post
347
348 mock_session_ctx = MagicMock()
349 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
350 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
351
352 with (
353 patch(
354 "maestro.muse_cli.commands.clone.store_pulled_commit",
355 new=AsyncMock(return_value=False),
356 ),
357 patch(
358 "maestro.muse_cli.commands.clone.store_pulled_object",
359 new=AsyncMock(return_value=False),
360 ),
361 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
362 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=hub),
363 ):
364 asyncio.run(
365 _clone_async(
366 url=url,
367 directory=str(target),
368 depth=1,
369 branch=None,
370 single_track=None,
371 no_checkout=False,
372 )
373 )
374
375 assert captured[0]["depth"] == 1
376
377
378 def test_clone_single_track_forwarded(tmp_path: pathlib.Path) -> None:
379 """--single-track TEXT is forwarded to Hub in the clone request."""
380 url = "https://hub.stori.app/repos/full-band"
381 target = tmp_path / "full-band"
382
383 captured: list[dict[str, object]] = []
384 mock_response = _make_clone_response()
385
386 hub = MagicMock()
387 hub.__aenter__ = AsyncMock(return_value=hub)
388 hub.__aexit__ = AsyncMock(return_value=None)
389
390 async def fake_post(path: str, **kwargs: object) -> MagicMock:
391 payload = kwargs.get("json", {})
392 if isinstance(payload, dict):
393 captured.append(payload)
394 return mock_response
395
396 hub.post = fake_post
397
398 mock_session_ctx = MagicMock()
399 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
400 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
401
402 with (
403 patch(
404 "maestro.muse_cli.commands.clone.store_pulled_commit",
405 new=AsyncMock(return_value=False),
406 ),
407 patch(
408 "maestro.muse_cli.commands.clone.store_pulled_object",
409 new=AsyncMock(return_value=False),
410 ),
411 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
412 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=hub),
413 ):
414 asyncio.run(
415 _clone_async(
416 url=url,
417 directory=str(target),
418 depth=None,
419 branch=None,
420 single_track="keys",
421 no_checkout=False,
422 )
423 )
424
425 assert captured[0]["single_track"] == "keys"
426
427
428 # ---------------------------------------------------------------------------
429 # Filesystem structure tests
430 # ---------------------------------------------------------------------------
431
432
433 def test_clone_writes_repo_id_from_hub(tmp_path: pathlib.Path) -> None:
434 """repo_id returned by Hub is written into .muse/repo.json."""
435 url = "https://hub.stori.app/repos/hub-id-test"
436 target = tmp_path / "hub-id-test"
437 response = _make_clone_response(repo_id="canonical-hub-uuid-9999")
438
439 _run_clone(url=url, directory=str(target), mock_response=response)
440
441 data = json.loads((target / ".muse" / "repo.json").read_text())
442 assert data["repo_id"] == "canonical-hub-uuid-9999"
443
444
445 def test_clone_updates_branch_ref_with_remote_head(tmp_path: pathlib.Path) -> None:
446 """After clone, .muse/refs/heads/<branch> contains the remote HEAD commit ID."""
447 url = "https://hub.stori.app/repos/ref-test"
448 target = tmp_path / "ref-test"
449 head_id = "commit-abc123def456" * 3
450 response = _make_clone_response(remote_head=head_id, default_branch="main")
451
452 _run_clone(url=url, directory=str(target), mock_response=response)
453
454 ref_path = target / ".muse" / "refs" / "heads" / "main"
455 assert ref_path.is_file()
456 assert ref_path.read_text(encoding="utf-8").strip() == head_id
457
458
459 def test_clone_writes_remote_tracking_head(tmp_path: pathlib.Path) -> None:
460 """After clone, .muse/remotes/origin/<branch> is written with remote HEAD."""
461 url = "https://hub.stori.app/repos/tracking-test"
462 target = tmp_path / "tracking-test"
463 head_id = "tracking-head-id99887766" * 2
464 response = _make_clone_response(remote_head=head_id, default_branch="main")
465
466 _run_clone(url=url, directory=str(target), mock_response=response)
467
468 stored = get_remote_head("origin", "main", target)
469 assert stored == head_id
470
471
472 def test_clone_no_checkout_skips_muse_work(tmp_path: pathlib.Path) -> None:
473 """--no-checkout leaves muse-work/ unpopulated."""
474 url = "https://hub.stori.app/repos/no-co-test"
475 target = tmp_path / "no-co-test"
476
477 _run_clone(url=url, directory=str(target), no_checkout=True)
478
479 assert not (target / "muse-work").is_dir()
480
481
482 def test_clone_without_no_checkout_creates_muse_work(tmp_path: pathlib.Path) -> None:
483 """Without --no-checkout, muse-work/ is created."""
484 url = "https://hub.stori.app/repos/co-test"
485 target = tmp_path / "co-test"
486
487 _run_clone(url=url, directory=str(target), no_checkout=False)
488
489 assert (target / "muse-work").is_dir()
490
491
492 def test_clone_branch_sets_head(tmp_path: pathlib.Path) -> None:
493 """--branch sets HEAD to the specified branch name."""
494 url = "https://hub.stori.app/repos/branch-test"
495 target = tmp_path / "branch-test"
496 response = _make_clone_response(default_branch="feature/guitar")
497
498 _run_clone(url=url, directory=str(target), branch="feature/guitar", mock_response=response)
499
500 head = (target / ".muse" / "HEAD").read_text(encoding="utf-8").strip()
501 assert head == "refs/heads/feature/guitar"
502
503
504 # ---------------------------------------------------------------------------
505 # DB storage tests
506 # ---------------------------------------------------------------------------
507
508
509 @pytest.mark.anyio
510 async def test_clone_stores_commits_in_db(muse_cli_db_session: object) -> None:
511 """Commits returned from Hub clone are stored via store_pulled_commit."""
512 from sqlalchemy.ext.asyncio import AsyncSession
513 from maestro.muse_cli.models import MuseCliCommit
514
515 session: AsyncSession = muse_cli_db_session # type: ignore[assignment]
516
517 commit_data: dict[str, object] = {
518 "commit_id": "cloned-commit-aabbcc" * 3,
519 "repo_id": "test-hub-repo",
520 "parent_commit_id": None,
521 "snapshot_id": "snap-clone-001",
522 "branch": "main",
523 "message": "Initial commit from Hub",
524 "author": "producer@example.com",
525 "committed_at": "2025-03-01T12:00:00+00:00",
526 "metadata": None,
527 }
528
529 inserted = await store_pulled_commit(session, commit_data)
530 await session.commit()
531
532 assert inserted is True
533
534 commit_id = str(commit_data["commit_id"])
535 stored = await session.get(MuseCliCommit, commit_id)
536 assert stored is not None
537 assert stored.branch == "main"
538 assert stored.message == "Initial commit from Hub"
539
540
541 # ---------------------------------------------------------------------------
542 # Hub error handling
543 # ---------------------------------------------------------------------------
544
545
546 def test_clone_hub_non_200_exits_3(tmp_path: pathlib.Path) -> None:
547 """Hub returning non-200 causes muse clone to exit 3 and cleans up the directory."""
548 import typer
549
550 url = "https://hub.stori.app/repos/error-test"
551 target = tmp_path / "error-test"
552
553 error_response = MagicMock()
554 error_response.status_code = 404
555 error_response.text = "Not found"
556
557 hub = _mock_hub(error_response)
558
559 mock_session_ctx = MagicMock()
560 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
561 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
562
563 with (
564 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
565 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=hub),
566 pytest.raises(typer.Exit) as exc_info,
567 ):
568 asyncio.run(
569 _clone_async(
570 url=url,
571 directory=str(target),
572 depth=None,
573 branch=None,
574 single_track=None,
575 no_checkout=False,
576 )
577 )
578
579 assert exc_info.value.exit_code == int(ExitCode.INTERNAL_ERROR)
580 # Partial directory must be cleaned up so retrying does not hit "already exists".
581 assert not target.exists()
582
583
584 def test_clone_network_error_exits_3(tmp_path: pathlib.Path) -> None:
585 """Network error during clone causes exit 3 and cleans up the directory."""
586 import httpx
587 import typer
588
589 url = "https://hub.stori.app/repos/net-err"
590 target = tmp_path / "net-err"
591
592 hub = MagicMock()
593 hub.__aenter__ = AsyncMock(return_value=hub)
594 hub.__aexit__ = AsyncMock(return_value=None)
595 hub.post = AsyncMock(side_effect=httpx.NetworkError("connection refused"))
596
597 mock_session_ctx = MagicMock()
598 mock_session_ctx.__aenter__ = AsyncMock(return_value=MagicMock())
599 mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
600
601 with (
602 patch("maestro.muse_cli.commands.clone.open_session", return_value=mock_session_ctx),
603 patch("maestro.muse_cli.commands.clone.MuseHubClient", return_value=hub),
604 pytest.raises(typer.Exit) as exc_info,
605 ):
606 asyncio.run(
607 _clone_async(
608 url=url,
609 directory=str(target),
610 depth=None,
611 branch=None,
612 single_track=None,
613 no_checkout=False,
614 )
615 )
616
617 assert exc_info.value.exit_code == int(ExitCode.INTERNAL_ERROR)
618 # Partial directory must be cleaned up so retrying does not hit "already exists".
619 assert not target.exists()
620
621
622 # ---------------------------------------------------------------------------
623 # CLI skeleton test (app registration)
624 # ---------------------------------------------------------------------------
625
626
627 def test_clone_registered_in_cli() -> None:
628 """'clone' command is registered in the muse CLI app."""
629 from typer.testing import CliRunner
630 from maestro.muse_cli.app import cli
631
632 runner = CliRunner()
633 result = runner.invoke(cli, ["--help"])
634 assert "clone" in result.output