roboco/tests/e2e_smoke/test_gitlab_live.py

"""Live-GitLab contract test for GitLabProvider (forge Phase 3).

Mirror of ``test_gitea_live.py``: fully self-seeding against a real GitLab
instance (gitlab.com works) — creates a uniquely-named private project under
the token's namespace, pushes real commits, and drives the provider end to
end: MR open → duplicate reshape → native list/filter → GitHub-shape
adaptation → diff reassembly → note review → commit-status CI reshape →
squash merge → branch delete → release → the oauth2 Basic-auth git-CLI
claim. The project is deleted afterwards (best-effort).

Skipped unless both env vars are set:

    ROBOCO_GITLAB_E2E_URL    e.g. https://gitlab.com
    ROBOCO_GITLAB_E2E_TOKEN  a PAT with `api` scope

Run tip: the token is a secret — export it in the shell, never write it
into a file or a compose entry.
"""

from __future__ import annotations

import base64
import os
import subprocess
import time
from typing import TYPE_CHECKING, Any
from urllib.parse import quote, urlsplit
from uuid import uuid4

import httpx
import pytest
from roboco.services.forge.base import RepoRef
from roboco.services.forge.gitlab import GitLabProvider

if TYPE_CHECKING:
    from pathlib import Path

_URL = os.environ.get("ROBOCO_GITLAB_E2E_URL", "")
_TOKEN = os.environ.get("ROBOCO_GITLAB_E2E_TOKEN", "")

pytestmark = [
    pytest.mark.asyncio,
    pytest.mark.skipif(
        not (_URL and _TOKEN),
        reason="ROBOCO_GITLAB_E2E_URL / ROBOCO_GITLAB_E2E_TOKEN not set",
    ),
]

_MERGE_SETTLE_ATTEMPTS = 20


def _split_url() -> tuple[str, str]:
    parts = urlsplit(_URL)
    host = parts.netloc or parts.path
    return (parts.scheme or "https"), host


def _api(method: str, path: str, **kwargs: object) -> httpx.Response:
    scheme, host = _split_url()
    return httpx.request(
        method,
        f"{scheme}://{host}/api/v4{path}",
        headers={"Authorization": f"Bearer {_TOKEN}"},
        timeout=30.0,
        **kwargs,  # type: ignore[arg-type]
    )


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


def _seed_project(tmp_path: Path, name: str) -> tuple[str, str]:
    """Create the project, push a feature commit; return (path, head_sha)."""
    resp = _api(
        "post",
        "/projects",
        json={
            "name": name,
            "visibility": "private",
            "initialize_with_readme": True,
            "default_branch": "main",
        },
    )
    assert resp.status_code == httpx.codes.CREATED, resp.text
    path_with_namespace = resp.json()["path_with_namespace"]
    scheme, host = _split_url()
    clone_url = f"{scheme}://oauth2:{_TOKEN}@{host}/{path_with_namespace}.git"
    clone = tmp_path / "clone"
    subprocess.run(
        ["git", "clone", clone_url, str(clone)],
        check=True,
        capture_output=True,
        text=True,
    )
    _git(clone, "config", "user.email", "e2e@example.com")
    _git(clone, "config", "user.name", "E2E")
    _git(clone, "config", "commit.gpgsign", "false")
    _git(clone, "checkout", "-b", "feat/e2e-change")
    (clone / "widget.txt").write_text("widget v2\n")
    _git(clone, "add", "widget.txt")
    _git(clone, "commit", "-m", "add widget")
    _git(clone, "push", "-u", "origin", "feat/e2e-change")
    return path_with_namespace, _git(clone, "rev-parse", "HEAD")


async def _verify_mr_flow(provider: GitLabProvider, ref: RepoRef, head_sha: str) -> int:
    """create → duplicate reshape → native filter → adapted object → diff →
    note review; returns the MR iid (as GitHub-shape ``number``)."""
    created = await provider.create_pr(
        ref,
        _TOKEN,
        head="feat/e2e-change",
        base="main",
        title="E2E change",
        body="live contract test",
    )
    assert created.is_success, created.text
    pr = created.json()
    pr_number: int = pr["number"]
    assert pr["html_url"]
    assert pr["head"]["ref"] == "feat/e2e-change"
    assert pr["base"]["ref"] == "main"

    duplicate = await provider.create_pr(
        ref,
        _TOKEN,
        head="feat/e2e-change",
        base="main",
        title="E2E change",
        body="dup",
    )
    assert duplicate.status_code == httpx.codes.UNPROCESSABLE_ENTITY, (
        duplicate.status_code
    )
    assert "already exists" in duplicate.text.lower()

    listed = await provider.list_pulls(ref, _TOKEN, head="feat/e2e-change", base="main")
    pulls = listed.json()
    assert [p["number"] for p in pulls] == [pr_number]
    assert pulls[0]["author_association"] == "NONE"

    fetched = (await provider.get_pr(ref, _TOKEN, pr_number)).json()
    assert fetched["head"]["sha"] == head_sha
    assert not fetched.get("merged")

    diff = await provider.get_pr_diff(ref, _TOKEN, pr_number)
    assert "widget.txt" in diff.text

    review = await provider.post_review(
        ref, _TOKEN, pr_number, body="looks fine", event="COMMENT"
    )
    assert review.is_success, review.text
    return pr_number


async def _verify_ci_reshapes(
    provider: GitLabProvider, ref: RepoRef, head_sha: str
) -> None:
    """A real commit status classifies through the check_runs reshape; the
    pipelines-based views stay well-shaped for a pipeline-less project."""
    status = _api(
        "post",
        f"/projects/{quote(ref.owner, safe='')}/statuses/{head_sha}",
        json={"state": "success", "context": "ci/e2e"},
    )
    assert status.status_code in (200, 201), status.text
    check_runs = (
        await provider.list_check_runs(ref, _TOKEN, head_sha, per_page=50)
    ).json()["check_runs"]
    assert check_runs and check_runs[0]["conclusion"] == "success"
    assert check_runs[0]["status"] == "completed"

    runs = (
        await provider.list_ci_runs(
            ref,
            _TOKEN,
            workflow=None,
            branch="feat/e2e-change",
            head_sha=None,
            per_page=5,
        )
    ).json()
    assert "workflow_runs" in runs
    workflows = (await provider.list_workflows(ref, _TOKEN, per_page=1)).json()
    assert "total_count" in workflows


async def _verify_merge_publish_cli(
    provider: GitLabProvider, ref: RepoRef, pr_number: int, scheme: str
) -> None:
    """Squash merge (retrying while GitLab computes mergeability) → merged
    flag → branch delete → release → oauth2 Basic-auth CLI claim."""
    merged_resp = None
    for _ in range(_MERGE_SETTLE_ATTEMPTS):
        merged_resp = await provider.merge_pr(
            ref, _TOKEN, pr_number, merge_method="squash"
        )
        if merged_resp.is_success:
            break
        time.sleep(1.0)
    assert merged_resp is not None
    assert merged_resp.is_success, merged_resp.text
    for _ in range(_MERGE_SETTLE_ATTEMPTS):
        if (await provider.get_pr(ref, _TOKEN, pr_number)).json().get("merged"):
            break
        time.sleep(0.5)
    assert (await provider.get_pr(ref, _TOKEN, pr_number)).json()["merged"] is True

    deleted = await provider.delete_branch_ref(ref, _TOKEN, "feat/e2e-change")
    assert deleted.status_code in (204, 200, 404), deleted.text

    release = await provider.create_release(
        ref,
        _TOKEN,
        tag_name="v0.0.1-e2e",
        name="v0.0.1-e2e",
        body="live",
        target_commitish="main",
    )
    assert release.status_code == httpx.codes.CREATED, release.text
    assert release.json().get("html_url")

    basic = base64.b64encode(f"oauth2:{_TOKEN}".encode()).decode()
    ls = subprocess.run(
        [
            "git",
            "-c",
            f"http.extraheader=Authorization: Basic {basic}",
            "ls-remote",
            f"{scheme}://{ref.host}/{ref.owner}.git",
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert ls.returncode == 0, ls.stderr
    assert "refs/heads/main" in ls.stdout


async def test_gitlab_live_contract(tmp_path: Path) -> None:
    scheme, host = _split_url()
    project_name = f"roboco-e2e-{uuid4().hex[:8]}"
    path_with_namespace, head_sha = _seed_project(tmp_path, project_name)
    ref = RepoRef(path_with_namespace, "", host=host)
    provider = GitLabProvider(host, scheme=scheme)

    parsed = provider.parse_repo_ref(f"{scheme}://{host}/{path_with_namespace}.git")
    assert parsed == ref

    repo_resp = await provider.get_repo(ref, _TOKEN)
    assert repo_resp.is_success, repo_resp.text
    repo_json: dict[str, Any] = repo_resp.json()
    assert repo_json["full_name"] == path_with_namespace
    assert "allow_squash_merge" in repo_json
    assert "allow_merge_commit" in repo_json

    pr_number = await _verify_mr_flow(provider, ref, head_sha)
    await _verify_ci_reshapes(provider, ref, head_sha)
    await _verify_merge_publish_cli(provider, ref, pr_number, scheme)

    _api("delete", f"/projects/{quote(path_with_namespace, safe='')}")