cc-ci/tests/gitea/custom/test_git_push.py

"""gitea — parity port of recipe-info/gitea/tests/git_push.py (phase gtea).

SOURCE: references/recipe-maintainer/recipe-info/gitea/tests/git_push.py

Original: create repo via API → clone → commit → push over HTTPS → verify commit via API → delete.
cc-ci port: same flow, adapted to the per-run domain and the harness admin creds from ops.py.
Non-vacuous: a broken SCM fails the push or the API verification.
"""

from __future__ import annotations

import base64
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from ops import admin_creds  # noqa: E402

import ssl

_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE


def _api(domain, path, method="GET", body=None, user="", password=""):
    data = json.dumps(body).encode() if body is not None else None
    auth = base64.b64encode(f"{user}:{password}".encode()).decode()
    headers = {"Authorization": f"Basic {auth}"}
    if data:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(
        f"https://{domain}/api/v1{path}", data=data, headers=headers, method=method
    )
    try:
        with urllib.request.urlopen(req, timeout=20, context=_CTX) as r:
            raw = r.read()
            return r.status, (json.loads(raw) if raw else {})
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw)
        except (ValueError, json.JSONDecodeError):
            return e.code, {}


def _run_git(args, cwd, env=None):
    result = subprocess.run(
        ["git"] + args,
        cwd=cwd,
        capture_output=True,
        text=True,
        env={**os.environ, **(env or {})},
        timeout=60,
    )
    if result.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} failed:\n{result.stderr}")
    return result.stdout.strip()


def test_git_push(live_app):
    """Parity with recipe-info/gitea/tests/git_push.py: create repo → clone → push → verify."""
    user, password = admin_creds(live_app)
    repo_name = "ci-test-push"

    # 1. Create test repo
    status, body = _api(
        live_app, "/user/repos", method="POST",
        body={"name": repo_name, "private": False, "auto_init": False},
        user=user, password=password,
    )
    assert status == 201, f"repo create HTTP {status}: {body}"
    clone_url = body.get("clone_url") or f"https://{live_app}/{user}/{repo_name}.git"

    tmpdir = tempfile.mkdtemp(prefix="ccci-gitea-push-")
    try:
        git_env = {
            "GIT_AUTHOR_NAME": "CI Test Bot",
            "GIT_AUTHOR_EMAIL": "ci@ci.local",
            "GIT_COMMITTER_NAME": "CI Test Bot",
            "GIT_COMMITTER_EMAIL": "ci@ci.local",
            # Embed credentials so HTTPS push works without interactive prompt.
            "GIT_CONFIG_COUNT": "1",
            "GIT_CONFIG_KEY_0": f"url.https://{user}:{password}@{live_app}/.insteadOf",
            "GIT_CONFIG_VALUE_0": f"https://{live_app}/",
        }

        # 2. Clone (empty repo)
        _run_git(["clone", clone_url, tmpdir], cwd="/tmp", env=git_env)
        _run_git(["checkout", "-b", "main"], cwd=tmpdir, env=git_env)

        # 3. Commit a file
        readme = os.path.join(tmpdir, "README.md")
        with open(readme, "w") as f:
            f.write(f"# {repo_name}\n\nAutomated ci push test.\n")
        _run_git(["add", "README.md"], cwd=tmpdir, env=git_env)
        _run_git(["commit", "-m", "test: automated push test"], cwd=tmpdir, env=git_env)

        # 4. Push
        _run_git(["push", "origin", "HEAD:main"], cwd=tmpdir, env=git_env)

        # 5. Verify commit landed via API
        status, commits = _api(
            live_app, f"/repos/{user}/{repo_name}/commits?limit=1",
            user=user, password=password,
        )
        assert status == 200 and commits, f"commit list HTTP {status}: {commits}"
        commit_msg = commits[0].get("commit", {}).get("message", "").strip()
        assert "automated push test" in commit_msg, (
            f"Unexpected commit message: {commit_msg!r}"
        )
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)
        # 6. Cleanup — delete the test repo
        _api(live_app, f"/repos/{user}/{repo_name}", method="DELETE", user=user, password=password)