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

"""gitea — LFS round-trip capstone (phase gtea, PR #1 lfs-plain-gitea).

This test is the PROOF for PR #1 (feat: support Git LFS on plain gitea):
  - On gitea main: compose.lfs.yml is ABSENT → test skips (declared EXPECTED_NA in PARITY.md).
  - On lfs-plain-gitea PR head: compose.lfs.yml is PRESENT → test runs and must pass.

What it tests:
  1. LFS object upload (git lfs push) → download round-trip: fetched bytes hash to the OID.
  2. LFS JWT secret stability: after `abra app restart`, the rendered app.ini LFS_JWT_SECRET
     is unchanged and tokens still validate (the regression the PR fixes).

Non-vacuous: a gitea without LFS enabled cannot accept lfs push (step 1 fails); a rotating
LFS_JWT_SECRET breaks existing tokens (step 2 fails).
"""

from __future__ import annotations

import base64
import hashlib
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__), ".."))
import ssl

from harness import lifecycle  # noqa: E402
from ops import admin_creds  # noqa: E402

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

_LFS_OVERLAY = "compose.lfs.yml"


def _lfs_available() -> bool:
    abra_dir = os.environ.get("ABRA_DIR") or os.path.expanduser("~/.abra")
    return os.path.exists(os.path.join(abra_dir, "recipes", "gitea", _LFS_OVERLAY))


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):
    full_env = {**os.environ, **(env or {})}
    result = subprocess.run(
        ["git"] + args, cwd=cwd, capture_output=True, text=True, env=full_env, timeout=120
    )
    if result.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} failed:\n{result.stderr}")
    return result.stdout.strip()


def test_lfs_roundtrip(live_app):
    """LFS object upload→download round-trip + JWT-secret stability after restart.

    Skips when compose.lfs.yml is absent (gitea main — LFS not enabled in this deploy).
    """
    if not _lfs_available():
        import pytest

        pytest.skip(
            "compose.lfs.yml absent in gitea recipe checkout — LFS is not enabled on this branch. "
            "This test runs on lfs-plain-gitea (PR #1) and is EXPECTED_NA on main."
        )

    user, password = admin_creds(live_app)
    repo_name = "ci-lfs-test"
    # Embed credentials directly in the URL (password is 32-char hex, URL-safe).
    cred_url = f"https://{user}:{password}@{live_app}/{user}/{repo_name}.git"
    git_env = {
        "GIT_AUTHOR_NAME": "CI LFS Bot",
        "GIT_AUTHOR_EMAIL": "ci@ci.local",
        "GIT_COMMITTER_NAME": "CI LFS Bot",
        "GIT_COMMITTER_EMAIL": "ci@ci.local",
        "GIT_SSL_NO_VERIFY": "true",
        "GIT_TERMINAL_PROMPT": "0",
    }

    # 1. Create LFS test repo
    status, body = _api(
        live_app,
        "/user/repos",
        method="POST",
        body={"name": repo_name, "private": False, "auto_init": True, "default_branch": "main"},
        user=user,
        password=password,
    )
    assert status in (201, 409), f"repo create HTTP {status}: {body}"

    tmpdir = tempfile.mkdtemp(prefix="ccci-gitea-lfs-")
    try:
        # 2. Clone repo
        _run_git(["clone", cred_url, tmpdir], cwd="/tmp", env=git_env)
        _run_git(["lfs", "install"], cwd=tmpdir, env=git_env)

        # 3. Track *.bin as LFS
        _run_git(["lfs", "track", "*.bin"], cwd=tmpdir, env=git_env)
        _run_git(["add", ".gitattributes"], cwd=tmpdir, env=git_env)

        # 4. Create a 1KB binary blob (content: random-looking but deterministic per test)
        blob = bytes(range(256)) * 4  # 1024 bytes
        blob_path = os.path.join(tmpdir, "testblob.bin")
        with open(blob_path, "wb") as f:
            f.write(blob)
        expected_sha256 = hashlib.sha256(blob).hexdigest()
        expected_oid = f"sha256:{expected_sha256}"

        _run_git(["add", "testblob.bin"], cwd=tmpdir, env=git_env)
        _run_git(["commit", "-m", "test: add LFS blob"], cwd=tmpdir, env=git_env)

        # 5. Push (LFS upload happens here)
        _run_git(["push", "origin", "HEAD:main"], cwd=tmpdir, env=git_env)

        # Verify git-lfs pointer was tracked correctly
        lfs_ls = subprocess.run(
            ["git", "lfs", "ls-files"],
            cwd=tmpdir,
            capture_output=True,
            text=True,
            env={**os.environ, **git_env},
        )
        assert (
            "testblob.bin" in lfs_ls.stdout
        ), f"testblob.bin not in git-lfs ls-files: {lfs_ls.stdout}"

        # 6. Download in a FRESH clone (proves the LFS server stores and serves the object)
        fresh_dir = tempfile.mkdtemp(prefix="ccci-gitea-lfs-dl-")
        try:
            _run_git(["clone", cred_url, fresh_dir], cwd="/tmp", env=git_env)
            fetched_path = os.path.join(fresh_dir, "testblob.bin")
            assert os.path.exists(fetched_path), "testblob.bin not fetched in fresh clone"
            with open(fetched_path, "rb") as f:
                fetched = f.read()
            fetched_sha256 = hashlib.sha256(fetched).hexdigest()
            assert (
                fetched_sha256 == expected_sha256
            ), f"LFS round-trip OID mismatch: expected {expected_oid}, got sha256:{fetched_sha256}"
        finally:
            shutil.rmtree(fresh_dir, ignore_errors=True)

        # 7. JWT-secret stability: restart gitea + assert LFS_JWT_SECRET unchanged.
        #    Read the current secret from inside the container's rendered app.ini.
        current_jwt = lifecycle.exec_in_app(
            live_app,
            ["sh", "-c", "grep -E '^LFS_JWT_SECRET' /etc/gitea/app.ini || echo NOT_FOUND"],
            timeout=30,
        ).strip()
        assert (
            current_jwt and "NOT_FOUND" not in current_jwt
        ), "Could not read LFS_JWT_SECRET from /etc/gitea/app.ini before restart"

        # Restart the gitea container
        lifecycle.exec_in_app(live_app, ["true"], timeout=5)  # no-op to confirm exec works
        subprocess.run(
            ["docker", "service", "update", "--force", live_app.replace(".", "_") + "_app"],
            capture_output=True,
            timeout=120,
        )
        # Wait for gitea to come back up

        # Re-read meta from the live_app fixture (meta is not in scope here — use the stored meta)
        import time

        deadline = time.time() + 120
        while time.time() < deadline:
            status2, _ = _api(live_app, "/version", user=user, password=password)
            if status2 == 200:
                break
            time.sleep(5)
        assert status2 == 200, "gitea did not come back up after restart"

        jwt_after = lifecycle.exec_in_app(
            live_app,
            ["sh", "-c", "grep -E '^LFS_JWT_SECRET' /etc/gitea/app.ini || echo NOT_FOUND"],
            timeout=30,
        ).strip()
        assert jwt_after == current_jwt, (
            "LFS_JWT_SECRET changed after restart — the regression the PR fixes is still present: "
            f"before={current_jwt!r} after={jwt_after!r}"
        )

        # 8. Verify a fresh clone still works after restart (tokens still validate)
        post_restart_dir = tempfile.mkdtemp(prefix="ccci-gitea-lfs-restart-")
        try:
            _run_git(["clone", cred_url, post_restart_dir], cwd="/tmp", env=git_env)
            pr_blob = os.path.join(post_restart_dir, "testblob.bin")
            assert os.path.exists(pr_blob), "testblob.bin not fetched in post-restart clone"
            with open(pr_blob, "rb") as f:
                pr_data = f.read()
            assert (
                hashlib.sha256(pr_data).hexdigest() == expected_sha256
            ), "LFS object corrupted after restart — JWT secret may have changed"
        finally:
            shutil.rmtree(post_restart_dir, ignore_errors=True)

    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)
        _api(live_app, f"/repos/{user}/{repo_name}", method="DELETE", user=user, password=password)