cc-ci/tests/gitea/ops.py

"""gitea — pre-op seed hooks (phase gtea).

Admin user setup:
  gitea has no default admin user after deploy. ops.pre_install creates a `ci_admin` user via
  the gitea CLI inside the container, stores the generated password in a per-domain temp file
  (/tmp/ccci-gitea-admin-<domain>.json, mode 600), and every subsequent pre_<op> hook reads it.
  The ops module is re-imported per op (the orchestrator does not cache it), so the file is the
  single durable credential store for the run duration.

Data-integrity marker:
  Marker = git repo named `ci-marker`, owned by `ci_admin`, with an auto-initialised README.md.
  pre_install  : create admin user + marker repo
  pre_upgrade  : assert marker exists (idempotent re-create as guard)
  pre_backup   : assert marker exists (idempotent re-create as guard)
  pre_restore  : DELETE the marker repo (diverge from backup state)
  test_upgrade : assert marker survived the chaos redeploy
  test_backup  : assert marker was captured by the backup
  test_restore : assert marker returned (restore reverted deletion)
"""

from __future__ import annotations

import base64
import json
import os
import secrets
import sys
import urllib.error
import urllib.request

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

_ADMIN_USER = "ci_admin"
_ADMIN_EMAIL = "ci@ci.local"
_MARKER_REPO = "ci-marker"

_SSL_CTX = None


def _ssl_ctx():
    global _SSL_CTX
    if _SSL_CTX is None:
        import ssl
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        _SSL_CTX = ctx
    return _SSL_CTX


def _creds_path(domain: str) -> str:
    return f"/tmp/ccci-gitea-admin-{domain}.json"


def _load_creds(domain: str) -> tuple[str, str] | None:
    path = _creds_path(domain)
    if os.path.exists(path):
        with open(path) as f:
            d = json.load(f)
        return d["user"], d["password"]
    return None


def _save_creds(domain: str, password: str) -> None:
    path = _creds_path(domain)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump({"user": _ADMIN_USER, "password": password}, f)


def _ensure_admin(domain: str) -> tuple[str, str]:
    """Return (user, password) for the ci_admin user, creating it if this is the first call."""
    existing = _load_creds(domain)
    if existing:
        return existing

    password = secrets.token_hex(16)
    try:
        lifecycle.exec_in_app(
            domain,
            [
                "gitea", "admin", "user", "create", "--admin",
                "--username", _ADMIN_USER,
                "--password", password,
                "--email", _ADMIN_EMAIL,
                "--must-change-password=false",
            ],
            timeout=120,
        )
    except RuntimeError as e:
        if "already exists" in str(e).lower():
            lifecycle.exec_in_app(
                domain,
                [
                    "gitea", "admin", "user", "change-password",
                    "--username", _ADMIN_USER,
                    "--password", password,
                ],
                timeout=60,
            )
        else:
            raise
    _save_creds(domain, password)
    return _ADMIN_USER, password


def _gitea_api(
    domain: str, path: str, method: str = "GET", body=None, user: str = "", password: str = ""
) -> tuple[int, object]:
    """Call the gitea REST API (basic-auth). Returns (status, json_body_or_None)."""
    data = json.dumps(body).encode() if body is not None else None
    auth = base64.b64encode(f"{user}:{password}".encode()).decode()
    headers: dict[str, str] = {"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=30, context=_ssl_ctx()) as r:
            raw = r.read()
            return r.status, (json.loads(raw) if raw else None)
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw)
        except (ValueError, json.JSONDecodeError):
            return e.code, None


def _create_marker_repo(domain: str, user: str, password: str) -> bool:
    """Create ci-marker repo with auto_init=True. Returns True if created or already exists."""
    status, _ = _gitea_api(
        domain, "/user/repos", method="POST",
        body={"name": _MARKER_REPO, "private": False, "auto_init": True, "default_branch": "main"},
        user=user, password=password,
    )
    return status in (201, 409)


def _delete_marker_repo(domain: str, user: str, password: str) -> bool:
    """Delete ci-marker repo. Returns True if deleted or already gone."""
    status, _ = _gitea_api(
        domain, f"/repos/{user}/{_MARKER_REPO}", method="DELETE",
        user=user, password=password,
    )
    return status in (204, 404)


def marker_repo_exists(domain: str, user: str, password: str) -> bool:
    """Check whether the ci-marker repo is present. Called by test_*.py overlays."""
    status, _ = _gitea_api(
        domain, f"/repos/{user}/{_MARKER_REPO}", user=user, password=password
    )
    return status == 200


def admin_creds(domain: str) -> tuple[str, str]:
    """Return (user, password) for the ci_admin account. Called by test_*.py overlays."""
    existing = _load_creds(domain)
    if existing:
        return existing
    raise RuntimeError(
        f"No admin creds for {domain} — was ops.pre_install called for this run?"
    )


def pre_install(ctx):
    """After deploy: create admin user + seed the marker repo."""
    # App is already deployed + healthy at this point (pre_install runs after deploy+healthcheck).
    # Wait explicitly so the API is fully ready (READY_PROBE guards this at the harness level, but
    # belt-and-suspenders here in case this op is called in isolation).
    generic.assert_serving(ctx.domain, ctx.meta)
    # Fresh install wiped the DB. Any creds file from a previous run on this domain is stale
    # (user no longer exists in the new DB). Remove it so _ensure_admin creates a fresh user.
    stale = _creds_path(ctx.domain)
    if os.path.exists(stale):
        os.remove(stale)
    user, password = _ensure_admin(ctx.domain)
    ok = _create_marker_repo(ctx.domain, user, password)
    assert ok, f"pre_install: could not create {_MARKER_REPO} repo on {ctx.domain}"
    print(f"  gitea ops: admin {user!r} + repo {_MARKER_REPO!r} ready on {ctx.domain}", flush=True)


def pre_upgrade(ctx):
    """Before upgrade: ensure marker repo exists (data-continuity baseline)."""
    user, password = _ensure_admin(ctx.domain)
    ok = _create_marker_repo(ctx.domain, user, password)
    assert ok, f"pre_upgrade: could not ensure {_MARKER_REPO} repo exists on {ctx.domain}"


def pre_backup(ctx):
    """Before backup: ensure marker repo exists (prove backup captures it)."""
    user, password = _ensure_admin(ctx.domain)
    ok = _create_marker_repo(ctx.domain, user, password)
    assert ok, f"pre_backup: could not ensure {_MARKER_REPO} repo exists on {ctx.domain}"


def pre_restore(ctx):
    """After backup, before restore: DELETE marker repo (diverge from backup state).
    A successful restore must bring it back; a no-op restore leaves it absent → test fails."""
    user, password = _ensure_admin(ctx.domain)
    # backupbot cycles the gitea container during backup — wait for it to be back up.
    generic.assert_serving(ctx.domain, ctx.meta)
    ok = _delete_marker_repo(ctx.domain, user, password)
    assert ok, f"pre_restore: could not delete {_MARKER_REPO} repo on {ctx.domain}"
    assert not marker_repo_exists(ctx.domain, user, password), (
        f"pre_restore: {_MARKER_REPO} still present after delete — divergence did not take"
    )
    print(f"  gitea ops: {_MARKER_REPO!r} deleted (diverged from backup state)", flush=True)