headquarter/apps/api/tests/integration/test_workspace_git.py

"""Integration tests for workspace git endpoints."""

import asyncio
import os
import subprocess
import tempfile
import uuid

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession

from src.models import GitRepository, Project, Workspace


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


def _get_user_id(client: TestClient) -> uuid.UUID:
    from src.auth.session import decode_session_cookie
    from src.config import Settings

    settings = Settings()
    cookie = client.cookies.get("session")
    data = decode_session_cookie(settings=settings, cookie_value=cookie)
    if not data:
        raise RuntimeError("No session")
    return uuid.UUID(data["user_id"])


@pytest.fixture
def test_repo_with_git_workspace(
    db_session: AsyncSession, authenticated_client: TestClient
):
    """Create a workspace backed by a real git repo with commits."""
    user_id = _get_user_id(authenticated_client)

    async def _create():
        project = Project(name="Git Test Project", owner_id=user_id)
        db_session.add(project)
        await db_session.flush()

        repo = GitRepository(
            name="git-test-repo",
            path="/tmp/git-test-repo",
            remote_url="https://example.com/repo.git",
            project_id=project.id,
            owner_id=user_id,
        )
        db_session.add(repo)
        await db_session.flush()

        tmpdir = tempfile.mkdtemp()
        _run_git("init", cwd=tmpdir)
        _run_git("config", "user.email", "test@test.com", cwd=tmpdir)
        _run_git("config", "user.name", "Test User", cwd=tmpdir)
        with open(os.path.join(tmpdir, "README.md"), "w") as f:
            f.write("# Test\n")
        _run_git("add", "README.md", cwd=tmpdir)
        _run_git("commit", "-m", "Initial commit", cwd=tmpdir)

        ws = Workspace(
            name="dev",
            repo_id=repo.id,
            user_id=user_id,
            branch="main",
            path=tmpdir,
        )
        db_session.add(ws)
        await db_session.commit()
        await db_session.refresh(ws)
        return ws

    return asyncio.run(_create())


class TestGitStatus:
    """Tests for GET /workspaces/{id}/git/status."""

    def test_status_clean(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        response = authenticated_client.get(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/status"
        )
        assert response.status_code == 200
        data = response.json()
        assert data["branch"] == "main"
        assert data["modified"] == []

    def test_status_modified(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        with open(os.path.join(test_repo_with_git_workspace.path, "README.md"), "a") as f:
            f.write("modified\n")

        response = authenticated_client.get(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/status"
        )
        assert response.status_code == 200
        assert "README.md" in response.json()["modified"]


class TestGitCommit:
    """Tests for POST /workspaces/{id}/git/commit."""

    def test_commit(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        with open(os.path.join(test_repo_with_git_workspace.path, "new.py"), "w") as f:
            f.write("print('hello')\n")

        response = authenticated_client.post(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/commit",
            json={"message": "Add new file"},
        )
        assert response.status_code == 200
        assert response.json()["status"] == "committed"

    def test_commit_missing_message(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        response = authenticated_client.post(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/commit",
            json={},
        )
        assert response.status_code == 400


class TestGitHistory:
    """Tests for GET /workspaces/{id}/git/history."""

    def test_history(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        response = authenticated_client.get(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/history"
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data["commits"]) >= 1
        assert data["commits"][0]["message"] == "Initial commit"


class TestGitCheckout:
    """Tests for POST /workspaces/{id}/git/checkout."""

    def test_checkout(
        self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
    ):
        _run_git("checkout", "-b", "feature", cwd=test_repo_with_git_workspace.path)
        _run_git("checkout", "main", cwd=test_repo_with_git_workspace.path)

        response = authenticated_client.post(
            f"/workspaces/{test_repo_with_git_workspace.id}/git/checkout",
            json={"branch": "feature"},
        )
        assert response.status_code == 200
        assert response.json()["branch"] == "feature"