469 lines
16 KiB
Python
469 lines
16 KiB
Python
"""Docker lifecycle and seeding helpers for the live-Gitea integration tests.
|
|
|
|
Everything here talks to the ``docker`` CLI through ``subprocess`` (no extra
|
|
Python dependency). The harness never builds images, never uses host bind
|
|
mounts and never prunes anything: it only removes the container and the
|
|
network it created itself, identified by a unique per-run id.
|
|
|
|
Secrets (admin password, tokens) are random per run, only ever live in memory,
|
|
and are excluded from ``repr`` and from every error message.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import atexit
|
|
import os
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from types import FrameType
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
# Single source of truth for the Gitea image (digest-pinned).
|
|
GITEA_IMAGE = (
|
|
"gitea/gitea:28.1.0@sha256:f505a0d3a41323b786e05942fc7a4075e07ae7c169fd9485649aa55ee543a21a"
|
|
)
|
|
NAME_PREFIX = "aegis-it-"
|
|
GITEA_PORT = 3000
|
|
ADMIN_LOGIN = "aegis-admin"
|
|
ADMIN_EMAIL = "aegis-admin@example.test"
|
|
READY_TIMEOUT_SECONDS = 90.0
|
|
NETWORK_MODE_ENV = "AEGIS_IT_NETWORK_MODE"
|
|
HTTP_TIMEOUT_SECONDS = 30.0
|
|
|
|
|
|
class HarnessError(RuntimeError):
|
|
"""Raised for harness failures. Messages never contain secrets."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# docker CLI helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_docker(
|
|
*args: str, check: bool = True, timeout: float = 120.0
|
|
) -> subprocess.CompletedProcess[str]:
|
|
"""Run ``docker <args>`` and return the completed process.
|
|
|
|
On failure the error names only the first two arguments (never the full
|
|
argv, which may carry a password) plus the command's stderr.
|
|
"""
|
|
try:
|
|
proc = subprocess.run(
|
|
["docker", *args],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except (FileNotFoundError, PermissionError) as exc:
|
|
raise HarnessError("the 'docker' CLI was not found or is not executable") from exc
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise HarnessError(f"docker {' '.join(args[:2])} timed out after {timeout}s") from exc
|
|
if check and proc.returncode != 0:
|
|
raise HarnessError(
|
|
f"docker {' '.join(args[:2])} failed (exit {proc.returncode}): {proc.stderr.strip()}"
|
|
)
|
|
return proc
|
|
|
|
|
|
def require_docker() -> None:
|
|
"""Fail with a clear message when the docker daemon is not usable."""
|
|
try:
|
|
proc = run_docker("version", "--format", "{{.Server.Version}}", check=False, timeout=30)
|
|
except HarnessError as exc:
|
|
raise HarnessError(
|
|
f"AEGIS_INTEGRATION=1 but Docker is unavailable: {exc}. "
|
|
"Install/start Docker or unset AEGIS_INTEGRATION."
|
|
) from exc
|
|
if proc.returncode != 0:
|
|
raise HarnessError(
|
|
"AEGIS_INTEGRATION=1 but the Docker daemon is not reachable "
|
|
f"({proc.stderr.strip() or 'no output'}). Start Docker or unset AEGIS_INTEGRATION."
|
|
)
|
|
|
|
|
|
def ensure_image(image: str = GITEA_IMAGE) -> None:
|
|
"""Pull the pinned image if it is not present locally."""
|
|
if run_docker("image", "inspect", image, check=False).returncode == 0:
|
|
return
|
|
run_docker("pull", image, timeout=600)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cleanup (idempotent, registered with atexit and signal handlers)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class RunResources:
|
|
"""Docker resources owned by one harness run."""
|
|
|
|
run_id: str
|
|
container: str
|
|
network: str
|
|
attached_container: str | None = None # job container joined in CI mode
|
|
cleaned: bool = False
|
|
|
|
|
|
_ACTIVE: dict[str, RunResources] = {}
|
|
_LOCK = threading.Lock()
|
|
_HOOKS_INSTALLED = False
|
|
_PREVIOUS_HANDLERS: dict[int, Any] = {}
|
|
|
|
|
|
def cleanup_resources(res: RunResources) -> None:
|
|
"""Remove this run's container and network (idempotent, best effort)."""
|
|
with _LOCK:
|
|
if res.cleaned:
|
|
return
|
|
res.cleaned = True
|
|
_ACTIVE.pop(res.run_id, None)
|
|
if res.attached_container:
|
|
run_docker("network", "disconnect", "-f", res.network, res.attached_container, check=False)
|
|
run_docker("rm", "-f", "-v", res.container, check=False)
|
|
run_docker("network", "rm", res.network, check=False)
|
|
|
|
|
|
def cleanup_all() -> None:
|
|
"""Clean every run still registered (atexit and signal path)."""
|
|
for res in list(_ACTIVE.values()):
|
|
try:
|
|
cleanup_resources(res)
|
|
except Exception: # noqa: BLE001 - best effort during interpreter exit
|
|
pass
|
|
|
|
|
|
def _signal_handler(signum: int, frame: FrameType | None) -> None:
|
|
cleanup_all()
|
|
previous = _PREVIOUS_HANDLERS.get(signum, signal.SIG_DFL)
|
|
signal.signal(signum, previous if previous is not None else signal.SIG_DFL)
|
|
if callable(previous):
|
|
previous(signum, frame)
|
|
else:
|
|
os.kill(os.getpid(), signum)
|
|
|
|
|
|
def _install_exit_hooks() -> None:
|
|
global _HOOKS_INSTALLED
|
|
if _HOOKS_INSTALLED:
|
|
return
|
|
_HOOKS_INSTALLED = True
|
|
atexit.register(cleanup_all)
|
|
if threading.current_thread() is not threading.main_thread():
|
|
return
|
|
for name in ("SIGTERM", "SIGHUP"):
|
|
signum = getattr(signal, name, None)
|
|
if signum is None:
|
|
continue
|
|
_PREVIOUS_HANDLERS[signum] = signal.getsignal(signum)
|
|
signal.signal(signum, _signal_handler)
|
|
|
|
|
|
def register_resources(res: RunResources) -> None:
|
|
"""Track a run so that atexit/signal handlers can clean it up."""
|
|
_install_exit_hooks()
|
|
with _LOCK:
|
|
_ACTIVE[res.run_id] = res
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gitea instance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def unique(prefix: str) -> str:
|
|
"""Return ``<prefix>-<8 hex>``, valid as a Gitea user, org or repo name."""
|
|
return f"{prefix}-{secrets.token_hex(4)}"
|
|
|
|
|
|
def _free_loopback_port() -> int:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return int(sock.getsockname()[1])
|
|
|
|
|
|
def _gitea_env(root_url: str) -> dict[str, str]:
|
|
"""Return the ``GITEA__section__KEY`` environment for the throwaway instance."""
|
|
return {
|
|
"GITEA__database__DB_TYPE": "sqlite3",
|
|
"GITEA__security__INSTALL_LOCK": "true",
|
|
"GITEA__security__SECRET_KEY": secrets.token_hex(32),
|
|
"GITEA__security__INTERNAL_TOKEN": secrets.token_hex(32),
|
|
"GITEA__server__ROOT_URL": root_url,
|
|
"GITEA__server__HTTP_PORT": str(GITEA_PORT),
|
|
"GITEA__server__DISABLE_SSH": "true",
|
|
"GITEA__server__OFFLINE_MODE": "true",
|
|
"GITEA__picture__DISABLE_GRAVATAR": "true",
|
|
"GITEA__oauth2__ENABLED": "true",
|
|
"GITEA__oauth2__JWT_SECRET": secrets.token_urlsafe(32),
|
|
"GITEA__actions__ENABLED": "true",
|
|
"GITEA__service__DISABLE_REGISTRATION": "true",
|
|
"GITEA__repository__ENABLE_PUSH_CREATE_USER": "false",
|
|
"GITEA__repository__ENABLE_PUSH_CREATE_ORG": "false",
|
|
"GITEA__mailer__ENABLED": "false",
|
|
"GITEA__webhook__ALLOWED_HOST_LIST": "*",
|
|
"GITEA__packages__ENABLED": "true",
|
|
"GITEA__api__MAX_RESPONSE_ITEMS": "50",
|
|
}
|
|
|
|
|
|
def _failure(response: httpx.Response, action: str) -> HarnessError:
|
|
return HarnessError(f"{action} failed: HTTP {response.status_code}: {response.text[:300]}")
|
|
|
|
|
|
@dataclass
|
|
class GiteaInstance:
|
|
"""A running throwaway Gitea plus helpers to seed and call it."""
|
|
|
|
base_url: str
|
|
admin_login: str
|
|
admin_password: str = field(repr=False)
|
|
admin_token: str = field(repr=False)
|
|
container: str
|
|
network: str
|
|
run_id: str
|
|
_client: httpx.Client = field(repr=False, default_factory=httpx.Client)
|
|
|
|
def api(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
token: str | None = None,
|
|
*,
|
|
anonymous: bool = False,
|
|
**kwargs: Any,
|
|
) -> httpx.Response:
|
|
"""Call ``/api/v1<path>``.
|
|
|
|
``token`` defaults to the admin token; pass ``anonymous=True`` for an
|
|
unauthenticated call, or ``auth=(login, password)`` for basic auth.
|
|
"""
|
|
headers: dict[str, str] = dict(kwargs.pop("headers", None) or {})
|
|
if not anonymous and "auth" not in kwargs:
|
|
headers["Authorization"] = f"token {token or self.admin_token}"
|
|
url = f"{self.base_url}/api/v1/{path.lstrip('/')}"
|
|
return self._client.request(method, url, headers=headers, **kwargs)
|
|
|
|
def unique(self, prefix: str) -> str:
|
|
"""Return ``<prefix>-<8 hex>``, valid as a Gitea user, org or repo name."""
|
|
return unique(prefix)
|
|
|
|
def create_user(self, login: str, admin: bool = False) -> tuple[str, str]:
|
|
"""Create a user with a random password and return ``(login, password)``."""
|
|
password = secrets.token_urlsafe(24)
|
|
response = self.api(
|
|
"POST",
|
|
"/admin/users",
|
|
json={
|
|
"username": login,
|
|
"email": f"{login}@example.test",
|
|
"password": password,
|
|
"must_change_password": False,
|
|
},
|
|
)
|
|
if response.status_code >= 300:
|
|
raise _failure(response, f"create user {login}")
|
|
if admin:
|
|
promote = self.api(
|
|
"PATCH",
|
|
f"/admin/users/{login}",
|
|
json={"login_name": login, "source_id": 0, "admin": True},
|
|
)
|
|
if promote.status_code >= 300:
|
|
raise _failure(promote, f"promote user {login}")
|
|
return login, password
|
|
|
|
def mint_token(self, login: str, password: str, scopes: list[str]) -> str:
|
|
"""Create an access token for ``login`` (basic auth) and return its secret."""
|
|
response = self.api(
|
|
"POST",
|
|
f"/users/{login}/tokens",
|
|
auth=(login, password),
|
|
json={"name": f"it-{secrets.token_hex(4)}", "scopes": scopes},
|
|
)
|
|
if response.status_code >= 300:
|
|
raise _failure(response, f"mint token for {login}")
|
|
token = response.json().get("sha1")
|
|
if not isinstance(token, str) or not token:
|
|
raise HarnessError(f"mint token for {login} returned no token")
|
|
return token
|
|
|
|
def create_org(self, name: str) -> dict[str, Any]:
|
|
"""Create an organization (as admin) and return its JSON."""
|
|
response = self.api("POST", "/orgs", json={"username": name})
|
|
if response.status_code >= 300:
|
|
raise _failure(response, f"create org {name}")
|
|
data: dict[str, Any] = response.json()
|
|
return data
|
|
|
|
def create_repo(
|
|
self, owner: str, name: str, auto_init: bool = True, org: bool = False
|
|
) -> dict[str, Any]:
|
|
"""Create a repository for a user or an org (as admin) and return its JSON."""
|
|
path = f"/orgs/{owner}/repos" if org else f"/admin/users/{owner}/repos"
|
|
response = self.api(
|
|
"POST",
|
|
path,
|
|
json={"name": name, "auto_init": auto_init, "default_branch": "main"},
|
|
)
|
|
if response.status_code >= 300:
|
|
raise _failure(response, f"create repo {owner}/{name}")
|
|
data: dict[str, Any] = response.json()
|
|
return data
|
|
|
|
def close(self) -> None:
|
|
"""Close the HTTP client."""
|
|
self._client.close()
|
|
|
|
|
|
def _container_logs_tail(container: str, lines: int = 100) -> str:
|
|
proc = run_docker("logs", "--tail", str(lines), container, check=False, timeout=30)
|
|
return (proc.stdout + proc.stderr).strip()
|
|
|
|
|
|
def _wait_ready(base_url: str, container: str) -> None:
|
|
deadline = time.monotonic() + READY_TIMEOUT_SECONDS
|
|
last_error = "no response"
|
|
while time.monotonic() < deadline:
|
|
state = run_docker(
|
|
"inspect", "--format", "{{.State.Running}}", container, check=False, timeout=15
|
|
)
|
|
if state.stdout.strip() == "false":
|
|
last_error = "container exited"
|
|
break
|
|
try:
|
|
response = httpx.get(f"{base_url}/api/v1/version", timeout=3.0)
|
|
if response.status_code == 200:
|
|
return
|
|
last_error = f"HTTP {response.status_code}"
|
|
except httpx.HTTPError as exc:
|
|
last_error = type(exc).__name__
|
|
time.sleep(1.0)
|
|
print(
|
|
f"--- gitea container logs (tail 100) ---\n{_container_logs_tail(container)}",
|
|
file=sys.stderr,
|
|
)
|
|
raise HarnessError(
|
|
f"Gitea at {base_url} was not ready within {READY_TIMEOUT_SECONDS:.0f}s ({last_error})"
|
|
)
|
|
|
|
|
|
def _start_container(res: RunResources, container_mode: bool) -> str:
|
|
"""Create the network, start Gitea and return the base URL the tests use."""
|
|
run_docker("network", "create", res.network)
|
|
last_error: HarnessError | None = None
|
|
attempts = 1 if container_mode else 3
|
|
for _ in range(attempts):
|
|
if container_mode:
|
|
base_url = f"http://{res.container}:{GITEA_PORT}"
|
|
publish: list[str] = []
|
|
else:
|
|
# ROOT_URL must match how tests reach Gitea, so the host port is chosen
|
|
# up front (loopback only) instead of letting Docker pick one.
|
|
port = _free_loopback_port()
|
|
base_url = f"http://127.0.0.1:{port}"
|
|
publish = ["-p", f"127.0.0.1:{port}:{GITEA_PORT}"]
|
|
env_args: list[str] = []
|
|
for key, value in _gitea_env(base_url).items():
|
|
env_args += ["-e", f"{key}={value}"]
|
|
try:
|
|
run_docker(
|
|
"run",
|
|
"-d",
|
|
"--name",
|
|
res.container,
|
|
"--network",
|
|
res.network,
|
|
"--label",
|
|
"aegis-it=1",
|
|
*publish,
|
|
*env_args,
|
|
GITEA_IMAGE,
|
|
)
|
|
except HarnessError as exc:
|
|
last_error = exc
|
|
run_docker("rm", "-f", "-v", res.container, check=False)
|
|
continue
|
|
if not container_mode:
|
|
mapped = run_docker("port", res.container, f"{GITEA_PORT}/tcp").stdout
|
|
if not any(line.strip().endswith(f":{port}") for line in mapped.splitlines()):
|
|
raise HarnessError("published port does not match the configured ROOT_URL")
|
|
return base_url
|
|
raise HarnessError(f"could not start the Gitea container: {last_error}")
|
|
|
|
|
|
def start_gitea() -> GiteaInstance:
|
|
"""Start a throwaway Gitea, seed the admin user and token, return the instance.
|
|
|
|
Raises ``HarnessError`` (after cleaning up) on any failure.
|
|
"""
|
|
require_docker()
|
|
ensure_image()
|
|
run_id = secrets.token_hex(4)
|
|
container_mode = os.environ.get(NETWORK_MODE_ENV, "").lower() == "container"
|
|
res = RunResources(
|
|
run_id=run_id, container=f"{NAME_PREFIX}{run_id}", network=f"{NAME_PREFIX}{run_id}"
|
|
)
|
|
register_resources(res)
|
|
try:
|
|
base_url = _start_container(res, container_mode)
|
|
if container_mode:
|
|
job_container = os.environ.get("HOSTNAME")
|
|
if not job_container:
|
|
raise HarnessError("AEGIS_IT_NETWORK_MODE=container requires $HOSTNAME")
|
|
run_docker("network", "connect", res.network, job_container)
|
|
res.attached_container = job_container
|
|
_wait_ready(base_url, res.container)
|
|
admin_password = secrets.token_urlsafe(24)
|
|
run_docker(
|
|
"exec",
|
|
"-u",
|
|
"git",
|
|
res.container,
|
|
"gitea",
|
|
"admin",
|
|
"user",
|
|
"create",
|
|
"--admin",
|
|
"--username",
|
|
ADMIN_LOGIN,
|
|
"--password",
|
|
admin_password,
|
|
"--email",
|
|
ADMIN_EMAIL,
|
|
"--must-change-password=false",
|
|
)
|
|
instance = GiteaInstance(
|
|
base_url=base_url,
|
|
admin_login=ADMIN_LOGIN,
|
|
admin_password=admin_password,
|
|
admin_token="",
|
|
container=res.container,
|
|
network=res.network,
|
|
run_id=run_id,
|
|
_client=httpx.Client(timeout=HTTP_TIMEOUT_SECONDS),
|
|
)
|
|
instance.admin_token = instance.mint_token(ADMIN_LOGIN, admin_password, ["all"])
|
|
return instance
|
|
except BaseException:
|
|
cleanup_resources(res)
|
|
raise
|
|
|
|
|
|
def stop_gitea(instance: GiteaInstance) -> None:
|
|
"""Close the client and remove the instance's Docker resources."""
|
|
instance.close()
|
|
with _LOCK:
|
|
res = _ACTIVE.get(instance.run_id)
|
|
if res is not None:
|
|
cleanup_resources(res)
|