"""WorkspaceService._sync_read_clone — the conventions read-clone refresh. The read clone is hard-reset to the default branch on every refresh. The bug it fixes: the old refresh used a token-less fetch, so a private repo's clone stayed frozen at clone-time and never saw commits merged afterwards. This proves the refresh actually advances the clone to a post-clone commit. """ from __future__ import annotations import base64 import subprocess from types import SimpleNamespace from typing import TYPE_CHECKING from unittest.mock import AsyncMock, MagicMock, patch import pytest from roboco.services import workspace as ws_mod from roboco.services.workspace import WorkspaceService if TYPE_CHECKING: from pathlib import Path def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", *args], cwd=str(cwd), capture_output=True, text=True, check=False ) def _commit(repo: Path, message: str) -> str: _git(repo, "add", "-A") _git(repo, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qm", message) return _git(repo, "rev-parse", "HEAD").stdout.strip() def test_sync_read_clone_advances_to_a_post_clone_commit(tmp_path: Path) -> None: origin = tmp_path / "origin" origin.mkdir() _git(origin, "init", "-q", "-b", "master") (origin / "README.md").write_text("v1\n") _commit(origin, "first") clone = tmp_path / "clone" _git(tmp_path, "clone", "-q", str(origin), str(clone)) first = _git(clone, "rev-parse", "HEAD").stdout.strip() # A new commit lands on origin AFTER the clone — the frozen-clone scenario. (origin / "NEW.txt").write_text("added later\n") second = _commit(origin, "second") assert first != second # token=None: a file:// fetch needs no auth (mirrors a public-repo refresh); # for a private https repo the token would be injected into the fetch URL. WorkspaceService._sync_read_clone(clone, f"file://{origin}", "master", None) assert _git(clone, "rev-parse", "HEAD").stdout.strip() == second assert (clone / "NEW.txt").is_file() def test_sync_read_clone_fetches_tags(tmp_path: Path) -> None: """The read clone must carry tags: the release manager derives "commits since last release" from the newest tag, and a tagless clone makes ``git describe`` fail so it walks the entire history (the 729-commit bug).""" origin = tmp_path / "origin" origin.mkdir() _git(origin, "init", "-q", "-b", "master") (origin / "README.md").write_text("v1\n") _commit(origin, "first") # The clone itself is tagless (--no-tags, as agent clones are) — the tag # lands on origin and the read-clone refresh must pull it in. clone = tmp_path / "clone" _git(tmp_path, "clone", "-q", "--no-tags", str(origin), str(clone)) assert _git(clone, "tag").stdout.strip() == "" # -c tag.gpgsign=false: create a lightweight tag regardless of a global # signing config that would otherwise force an annotated (signed) tag. _git(origin, "-c", "tag.gpgsign=false", "tag", "v0.17.0") WorkspaceService._sync_read_clone(clone, f"file://{origin}", "master", None) assert "v0.17.0" in _git(clone, "tag").stdout.split() assert _git(clone, "describe", "--tags", "--abbrev=0").stdout.strip() == "v0.17.0" def test_sync_read_clone_is_best_effort_on_unreachable_origin(tmp_path: Path) -> None: origin = tmp_path / "origin" origin.mkdir() _git(origin, "init", "-q", "-b", "master") (origin / "README.md").write_text("v1\n") _commit(origin, "first") clone = tmp_path / "clone" _git(tmp_path, "clone", "-q", str(origin), str(clone)) head = _git(clone, "rev-parse", "HEAD").stdout.strip() # A bogus origin must not raise — the refresh logs and leaves the clone as-is. WorkspaceService._sync_read_clone(clone, "file:///nonexistent/repo", "master", None) assert _git(clone, "rev-parse", "HEAD").stdout.strip() == head def test_sync_read_clone_with_token_uses_extraheader_not_url(tmp_path: Path) -> None: """A private-repo refresh injects the PAT via ``-c http.extraheader``, never URL-embedded into the fetch argv (mirrors the clone PAT-leak fix — H11).""" clone = tmp_path / "clone" clone.mkdir() token = "ghp_SECRETARGV" git_url = "https://github.com/o/r" captured: list[list[str]] = [] def _fake_run(argv: list[str], **_kw: object) -> subprocess.CompletedProcess[str]: captured.append(list(argv)) return subprocess.CompletedProcess(argv, 0, "", "") with patch("roboco.services.workspace.subprocess.run", side_effect=_fake_run): WorkspaceService._sync_read_clone(clone, git_url, "master", token) fetch_argv = next(a for a in captured if "fetch" in a) # Token never appears in argv... assert token not in fetch_argv assert f"https://{token}@" not in fetch_argv # ...the bare URL is the fetch ref... assert git_url in fetch_argv # ...and the basic-auth extraheader carries the encoded token. expected = base64.b64encode(f"x-access-token:{token}".encode()).decode() assert f"http.extraheader=Authorization: Basic {expected}" in fetch_argv # --- ensure_read_clone force-refetch (M45) ------------------------------------- # The 30s refresh TTL on the conventions read clone means a conventions commit # that merges to the default branch within 30s leaves the clone's HEAD stale, # so the sha-keyed conventions cache serves a stale map. `force=True` bypasses # the TTL so conventions reads always see the current default-branch HEAD. def _read_clone_service(tmp_path: Path) -> WorkspaceService: svc = WorkspaceService(MagicMock()) object.__setattr__(svc, "root", tmp_path) return svc def _read_clone_workspace(svc: WorkspaceService, slug: str) -> Path: workspace = svc.root / slug / "_meta" / "conventions" (workspace / ".git" / "objects").mkdir(parents=True) (workspace / ".git" / "HEAD").write_text("ref: refs/heads/master\n") return workspace @pytest.mark.asyncio async def test_ensure_read_clone_force_bypasses_ttl(tmp_path: Path) -> None: """force=True fetches even when the workspace was just synced (within TTL).""" svc = _read_clone_service(tmp_path) slug = "p" workspace = _read_clone_workspace(svc, slug) project = SimpleNamespace( slug=slug, default_branch="master", git_url="https://example/o/r" ) project_service = MagicMock() project_service.get_by_slug = AsyncMock(return_value=project) fetched: list[Path] = [] def _fake_sync(root: Path, *_a: object, **_k: object) -> None: fetched.append(root) # Pre-seed the synced map to "just now" so the TTL branch would skip. ws_mod._read_clone_synced[str(workspace)] = ws_mod._monotonic() try: with ( patch( "roboco.services.project.get_project_service", return_value=project_service, ), patch.object(WorkspaceService, "_is_workspace_healthy", return_value=True), patch.object( WorkspaceService, "_read_clone_token", new=staticmethod(AsyncMock(return_value=None)), ), patch.object(WorkspaceService, "_prune_broken_refs", return_value=None), patch.object(WorkspaceService, "_sync_read_clone", side_effect=_fake_sync), ): await svc.ensure_read_clone(slug, force=True) finally: ws_mod._read_clone_synced.pop(str(workspace), None) assert fetched, "force=True must fetch even within the 30s TTL" @pytest.mark.asyncio async def test_ensure_read_clone_default_respects_ttl(tmp_path: Path) -> None: """Default force=False skips the fetch when the workspace was just synced.""" svc = _read_clone_service(tmp_path) slug = "p" workspace = _read_clone_workspace(svc, slug) project = SimpleNamespace( slug=slug, default_branch="master", git_url="https://example/o/r" ) project_service = MagicMock() project_service.get_by_slug = AsyncMock(return_value=project) fetched: list[Path] = [] def _fake_sync(root: Path, *_a: object, **_k: object) -> None: fetched.append(root) ws_mod._read_clone_synced[str(workspace)] = ws_mod._monotonic() try: with ( patch( "roboco.services.project.get_project_service", return_value=project_service, ), patch.object(WorkspaceService, "_is_workspace_healthy", return_value=True), patch.object( WorkspaceService, "_read_clone_token", new=staticmethod(AsyncMock(return_value=None)), ), patch.object(WorkspaceService, "_prune_broken_refs", return_value=None), patch.object(WorkspaceService, "_sync_read_clone", side_effect=_fake_sync), ): await svc.ensure_read_clone(slug) finally: ws_mod._read_clone_synced.pop(str(workspace), None) assert not fetched, "default force=False must skip the fetch within the TTL"