roboco/tests/integration/test_git_conventions_pr.py

"""GitService.open_conventions_pr commits the file locally; PR is best-effort."""

from __future__ import annotations

import subprocess
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock
from uuid import uuid4

from roboco.config import settings
from roboco.db.tables import AgentTable, ProjectTable
from roboco.models import AgentRole, AgentStatus, Team
from roboco.services.forge import RepoRef
from roboco.services.git import _ConventionsPr, get_git_service

if TYPE_CHECKING:
    from pathlib import Path

    import pytest
    from sqlalchemy.ext.asyncio import AsyncSession

_SCAFFOLD_BRANCH = "chore/roboco-conventions-scaffold"


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


async def _seed_project(
    db: AsyncSession, workspace_path: str, *, slug: str | None = None
) -> ProjectTable:
    agent = AgentTable(
        id=uuid4(),
        name="Dev",
        slug=f"be-dev-{uuid4().hex[:8]}",
        role=AgentRole.DEVELOPER,
        team=Team.BACKEND,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="dev",
        capabilities=[],
        permissions={},
        metrics={},
    )
    db.add(agent)
    await db.flush()
    project = ProjectTable(
        id=uuid4(),
        name="G-Proj",
        slug=slug or f"g-proj-{uuid4().hex[:8]}",
        git_url="https://example.com/r.git",
        default_branch="master",
        assigned_cell=Team.BACKEND,
        created_by=agent.id,
        workspace_path=workspace_path,
    )
    db.add(project)
    await db.flush()
    return project


async def test_open_conventions_pr_commits_locally_without_remote(
    db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    # open_conventions_pr only accepts a workspace_path under
    # {workspaces_root}/{slug} (the containment guard), so anchor the root at
    # the test dir and place the repo under the project's own slug.
    monkeypatch.setattr(settings, "workspaces_root", str(tmp_path))
    slug = f"g-proj-{uuid4().hex[:8]}"
    repo = tmp_path / slug / "repo"
    repo.mkdir(parents=True)
    _git(repo, "init", "-b", "master")
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")
    _git(repo, "config", "commit.gpgsign", "false")
    (repo / "README.md").write_text("# r\n")
    _git(repo, "add", "README.md")
    _git(repo, "commit", "-m", "init")

    project = await _seed_project(db_session, str(repo), slug=slug)
    git = get_git_service(db_session)
    result = await git.open_conventions_pr(
        project.slug,
        content="version: 1\n",
        title="scaffold",
        body="b",
    )

    assert result is not None
    assert result["pr_number"] is None  # no git token / remote → PR not opened
    show = subprocess.run(
        ["git", "show", f"{_SCAFFOLD_BRANCH}:.roboco/conventions.yml"],
        cwd=repo,
        capture_output=True,
        text=True,
        check=False,
    )
    assert show.returncode == 0
    assert show.stdout == "version: 1\n"


async def test_open_conventions_pr_returns_none_without_workspace(
    db_session: AsyncSession, tmp_path: Path
) -> None:
    project = await _seed_project(db_session, str(tmp_path / "does-not-exist"))
    git = get_git_service(db_session)
    result = await git.open_conventions_pr(
        project.slug,
        content="version: 1\n",
        title="t",
        body="b",
    )
    assert result is None


async def test_open_conventions_pr_force_pushes_scaffold_branch(
    db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    # Second call: scaffold branch already on remote (non-fast-forward).
    # Must force-push (force=True) and return a real pr_number, not the
    # unopened {"pr_number": None} dict.
    repo = tmp_path / "repo"
    repo.mkdir()
    _git(repo, "init", "-b", "master")
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "T")
    _git(repo, "config", "commit.gpgsign", "false")
    (repo / "README.md").write_text("# r\n")
    _git(repo, "add", "README.md")
    _git(repo, "commit", "-m", "init")

    project = await _seed_project(db_session, str(repo))
    git = get_git_service(db_session)

    pushed: list[bool] = []

    async def _fake_push(
        _workspace: Any, force: bool = False, _branch: str | None = None
    ) -> tuple[str, int]:
        pushed.append(force)
        return (_SCAFFOLD_BRANCH, 1)

    async def _fake_token(_slug: str) -> str:
        return "tok"

    monkeypatch.setattr(git, "_token_for_project", _fake_token)
    monkeypatch.setattr(git, "push", _fake_push)
    monkeypatch.setattr(
        git, "_parse_github_remote", lambda _ws: RepoRef("owner", "repo")
    )

    _pr_number = 42
    _pr_url = "https://github.com/owner/repo/pull/42"

    class _Resp:
        is_success = True

        def json(self) -> dict[str, object]:
            return {"number": _pr_number, "html_url": _pr_url}

    async def _fake_post_pr(_repo_ref: RepoRef, _token: str, _body: Any) -> _Resp:
        return _Resp()

    monkeypatch.setattr(git, "_post_pr", _fake_post_pr)
    monkeypatch.setattr(git, "_apply_pr_labels", AsyncMock())

    spec = _ConventionsPr(
        content="version: 1\n",
        branch=_SCAFFOLD_BRANCH,
        title="scaffold",
        body="b",
    )
    result = await git._push_and_open_conventions_pr(project.slug, repo, "master", spec)

    assert pushed == [True], "second call must force-push (force=True)"
    assert result["pr_number"] == _pr_number
    assert result["pr_url"] == _pr_url