roboco/tests/integration/services/test_dep_update_probe.py

"""dry_upgrade_changes_lockfile — the read-only lockfile-diff probe.

Runs the project's dep_update_command in an isolated clone of the read clone and
reports whether a lockfile path got dirty — without ever mutating the read clone
or committing/pushing. Fail-safe: a null/failing command returns False.
"""

from __future__ import annotations

import asyncio
import subprocess
import time
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock

import pytest
from roboco.services.workspace import WorkspaceService, _ensure_lock_for

if TYPE_CHECKING:
    from pathlib import Path


def _git(cwd: Path, *args: str) -> None:
    subprocess.run(
        ["git", *args], cwd=str(cwd), capture_output=True, text=True, check=True
    )


def _make_read_clone(tmp_path: Path) -> Path:
    repo = tmp_path / "readclone"
    repo.mkdir()
    _git(repo, "init", "-q")
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "t")
    (repo / "uv.lock").write_text("version = 1\n")
    _git(repo, "add", "-A")
    _git(repo, "commit", "-q", "-m", "init")
    return repo


def _svc(read_clone: Path) -> Any:
    svc: Any = WorkspaceService.__new__(WorkspaceService)
    svc.ensure_read_clone = AsyncMock(return_value=read_clone)
    return svc


def _project(command: str | None, paths: list[str] | None = None) -> MagicMock:
    return MagicMock(slug="p", dep_update_command=command, dep_update_paths=paths)


@pytest.mark.asyncio
async def test_dirtying_a_lockfile_returns_true(tmp_path: Path) -> None:
    read_clone = _make_read_clone(tmp_path)
    svc = _svc(read_clone)
    cmd = "python3 -c \"open('uv.lock','a').write('x')\""

    assert await svc.dry_upgrade_changes_lockfile(_project(cmd)) is True

    # The read clone itself is never mutated by the probe.
    status = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=str(read_clone),
        capture_output=True,
        text=True,
        check=True,
    )
    assert status.stdout.strip() == ""


@pytest.mark.asyncio
async def test_noop_command_returns_false(tmp_path: Path) -> None:
    svc = _svc(_make_read_clone(tmp_path))
    assert (
        await svc.dry_upgrade_changes_lockfile(_project('python3 -c "pass"')) is False
    )


@pytest.mark.asyncio
async def test_null_command_returns_false(tmp_path: Path) -> None:
    svc = _svc(_make_read_clone(tmp_path))
    assert await svc.dry_upgrade_changes_lockfile(_project(None)) is False


@pytest.mark.asyncio
async def test_failing_command_returns_false(tmp_path: Path) -> None:
    svc = _svc(_make_read_clone(tmp_path))
    cmd = 'python3 -c "import sys; sys.exit(1)"'
    assert await svc.dry_upgrade_changes_lockfile(_project(cmd)) is False


@pytest.mark.asyncio
async def test_explicit_dep_update_paths_scope(tmp_path: Path) -> None:
    read_clone = _make_read_clone(tmp_path)
    svc = _svc(read_clone)
    # Command dirties uv.lock, but we only watch a different lockfile → False.
    cmd = "python3 -c \"open('uv.lock','a').write('x')\""
    project = _project(cmd, paths=["pnpm-lock.yaml"])
    assert await svc.dry_upgrade_changes_lockfile(project) is False


@pytest.mark.asyncio
async def test_probe_holds_read_clone_lock_across_local_clone(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The dep-update probe holds the read-clone lock across the local
    ``git clone --local`` so a concurrent ``_sync_read_clone`` cannot mutate the
    read clone mid-clone; released before the upgrade (which runs on an
    independent copy)."""
    read_clone = _make_read_clone(tmp_path)
    svc = _svc(read_clone)
    # Unique slug → a fresh lock not shared with any other test.
    project = MagicMock(
        slug="f116-probe", dep_update_command="python3 -c pass", dep_update_paths=None
    )
    lock = _ensure_lock_for("f116-probe", "_meta-conventions")
    assert not lock.locked()

    def slow_clone(read_clone: Path, clone_dir: Path, timeout: float) -> None:
        # Real local clone so the dir is valid, then hold the lock a while so
        # the test coroutine can observe the lock is held mid-clone.
        subprocess.run(
            [
                "git",
                "clone",
                "--local",
                "--no-hardlinks",
                str(read_clone),
                str(clone_dir),
            ],
            capture_output=True,
            text=True,
            timeout=timeout,
            check=True,
        )
        time.sleep(0.4)

    def noop_probe(_clone_dir: Path, _command: str, _lock_paths: list[str]) -> bool:
        return False

    monkeypatch.setattr(WorkspaceService, "_clone_local_into", staticmethod(slow_clone))
    monkeypatch.setattr(
        WorkspaceService, "_probe_lockfile_on_clone", staticmethod(noop_probe)
    )

    probe_task = asyncio.create_task(svc.dry_upgrade_changes_lockfile(project))
    # ensure_read_clone is mocked (instant) → the probe immediately enters the
    # lock + clone step. Give it a beat, then assert the lock is held.
    await asyncio.sleep(0.1)
    assert lock.locked(), "read-clone lock must be held during the local-clone step"
    await asyncio.wait_for(probe_task, timeout=3)
    assert not lock.locked(), "lock released after the clone step (upgrade needs none)"