diff --git a/tests/integration/README.md b/tests/integration/README.md new file mode 100644 index 0000000..5742091 --- /dev/null +++ b/tests/integration/README.md @@ -0,0 +1,52 @@ +# Live-Gitea integration tests + +These tests run against a throwaway Gitea 28.1.0 started in Docker. They are +opt-in: without `AEGIS_INTEGRATION=1` every test here is skipped and Docker is +never touched, so the default `uv run pytest` stays fast. + +```bash +AEGIS_INTEGRATION=1 uv run pytest tests/integration -m integration -v +``` + +An integration-only run covers little of `src/`, so the unit-test coverage gate +is lowered to 0 for runs that collect nothing but `tests/integration`. If Docker +is unavailable while `AEGIS_INTEGRATION=1` is set, the session fails with a clear +message instead of skipping. + +## What the harness does + +- Image: `GITEA_IMAGE` in `harness.py` (digest-pinned, single constant). +- One container and one network per run, both named `aegis-it-<8 hex>`. + No host bind mounts, no image builds, no `docker prune`. +- Gitea is configured through `GITEA__section__KEY` env vars (sqlite3, + `INSTALL_LOCK`, random `SECRET_KEY`/`INTERNAL_TOKEN`/OAuth2 JWT secret, + Actions, packages and OAuth2 enabled, mailer off, webhook allowed host list + `*`, `MAX_RESPONSE_ITEMS=50`). +- Readiness: polls `GET /api/v1/version` for up to 90 s. On failure the last 100 + log lines of the container are written to stderr. +- Seeding: admin `aegis-admin` (random password) via `gitea admin user create`, + plus a full-scope token. All data is fake and random per run. Passwords and + tokens are kept out of `repr` and error messages. +- Cleanup: `docker rm -f -v ` and `docker network rm ` from the + session fixture finalizer, an `atexit` hook, and SIGTERM/SIGHUP handlers. + +## Reachability modes + +1. Local (default): a free loopback port is chosen, published as + `127.0.0.1::3000`, and `ROOT_URL` is set to match. Base URL + `http://127.0.0.1:`. +2. CI (`AEGIS_IT_NETWORK_MODE=container`): for test runs inside a job container + that shares the host Docker socket. No port is published. Gitea joins the + per-run network, the job container (`$HOSTNAME`) is connected to the same + network with `docker network connect`, and tests reach Gitea at + `http://aegis-it-:3000`. On cleanup the job container is disconnected + again before the network is removed. + +## Fixtures and helpers + +- `gitea` (session): a `GiteaInstance` with `base_url`, `admin_login`, + `admin_token`, `api(method, path, token=None, **kw)` (defaults to the admin + token, `anonymous=True` sends no credentials), `create_user`, `mint_token`, + `create_org`, `create_repo`, `unique`. +- `regular_user` (session): non-admin user with its own full-scope token. +- `unique_name` (function): a unique name derived from the test name. diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..2267438 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1 @@ +"""Live-Gitea integration tests (opt-in, see README.md).""" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..f9d39d2 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,99 @@ +"""Opt-in fixtures for the live-Gitea integration tests. + +Tests in this directory are marked ``integration`` and skipped unless +``AEGIS_INTEGRATION=1``. The default test run never touches Docker. +""" + +from __future__ import annotations + +import os +import re +from collections.abc import Iterator +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +from tests.integration.harness import ( + GiteaInstance, + HarnessError, + start_gitea, + stop_gitea, + unique, +) + +_HERE = Path(__file__).parent.resolve() +ENABLE_ENV = "AEGIS_INTEGRATION" + + +def integration_enabled() -> bool: + """Return True when the opt-in env var is set.""" + return os.environ.get(ENABLE_ENV) == "1" + + +def pytest_configure(config: pytest.Config) -> None: + """Register the ``integration`` marker (idempotent with pyproject).""" + config.addinivalue_line( + "markers", "integration: live Gitea in Docker, opt-in via AEGIS_INTEGRATION=1" + ) + + +def _is_integration(item: pytest.Item) -> bool: + try: + return _HERE in Path(str(item.path)).resolve().parents + except (OSError, ValueError): # pragma: no cover - defensive + return False + + +def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: + """Mark and (unless opted in) skip every test under tests/integration.""" + selected = [item for item in items if _is_integration(item)] + skip = pytest.mark.skip(reason=f"live Gitea tests need {ENABLE_ENV}=1 and Docker") + for item in selected: + item.add_marker(pytest.mark.integration) + if not integration_enabled(): + item.add_marker(skip) + if selected and len(selected) == len(items): + # An integration-only run covers almost none of src/, so the unit-test + # coverage gate does not apply to it. + # pytest-cov reads its options from the early-config namespace. + for namespace in (config.option, config.known_args_namespace): + if getattr(namespace, "cov_fail_under", None) is not None: + namespace.cov_fail_under = 0 + + +@pytest.fixture(scope="session") +def gitea() -> Iterator[GiteaInstance]: + """A throwaway Gitea in Docker, shared by the whole session.""" + try: + instance = start_gitea() + except HarnessError as exc: + pytest.fail(f"integration harness could not start Gitea: {exc}", pytrace=False) + try: + yield instance + finally: + stop_gitea(instance) + + +@pytest.fixture +def unique_name(request: pytest.FixtureRequest) -> str: + """A unique, Gitea-safe name derived from the test name.""" + slug = re.sub(r"[^a-z0-9]+", "-", request.node.name.lower()).strip("-")[:16].strip("-") + return unique(slug or "t") + + +@dataclass +class RegularUser: + """A non-admin user with its own full-scope token.""" + + login: str + password: str = field(repr=False) + token: str = field(repr=False) + + +@pytest.fixture(scope="session") +def regular_user(gitea: GiteaInstance) -> RegularUser: + """A non-admin user plus a full-scope token minted for that user.""" + login, password = gitea.create_user(gitea.unique("it-user")) + token = gitea.mint_token(login, password, ["all"]) + return RegularUser(login=login, password=password, token=token) diff --git a/tests/integration/harness.py b/tests/integration/harness.py new file mode 100644 index 0000000..25fc2cc --- /dev/null +++ b/tests/integration/harness.py @@ -0,0 +1,468 @@ +"""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 `` 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 ``-<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``. + + ``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 ``-<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) diff --git a/tests/integration/test_harness.py b/tests/integration/test_harness.py new file mode 100644 index 0000000..9ce3c2e --- /dev/null +++ b/tests/integration/test_harness.py @@ -0,0 +1,140 @@ +"""Smoke tests of the integration harness itself (needs Docker, opt-in).""" + +from __future__ import annotations + +import pytest + +from tests.integration import harness +from tests.integration.conftest import RegularUser +from tests.integration.harness import GiteaInstance, HarnessError, RunResources + + +def _names(kind: str, run_id: str) -> str: + listing = { + "container": ["ps", "-a", "--filter", f"name={harness.NAME_PREFIX}{run_id}"], + "network": ["network", "ls", "--filter", f"name={harness.NAME_PREFIX}{run_id}"], + }[kind] + fmt = "{{.Names}}" if kind == "container" else "{{.Name}}" + return harness.run_docker(*listing, "--format", fmt).stdout.strip() + + +def test_image_is_digest_pinned() -> None: + assert "@sha256:" in harness.GITEA_IMAGE + assert harness.GITEA_IMAGE.startswith("gitea/gitea:28.1.0@") + + +def test_version_endpoint_reachable(gitea: GiteaInstance) -> None: + response = gitea.api("GET", "/version", anonymous=True) + assert response.status_code == 200 + assert response.json()["version"].startswith("28.1.0") + + +def test_admin_token_is_admin(gitea: GiteaInstance) -> None: + response = gitea.api("GET", "/user") + assert response.status_code == 200 + body = response.json() + assert body["login"] == gitea.admin_login + assert body["is_admin"] is True + + +def test_admin_token_has_all_scope(gitea: GiteaInstance) -> None: + response = gitea.api("GET", "/token") + assert response.status_code == 200 + assert "all" in response.json()["scopes"] + + +def test_regular_user_is_not_admin(gitea: GiteaInstance, regular_user: RegularUser) -> None: + me = gitea.api("GET", "/user", token=regular_user.token) + assert me.status_code == 200 + assert me.json()["login"] == regular_user.login + assert me.json()["is_admin"] is False + forbidden = gitea.api("GET", "/admin/users", token=regular_user.token) + assert forbidden.status_code == 403 + + +def test_anonymous_call_is_rejected(gitea: GiteaInstance) -> None: + assert gitea.api("GET", "/user", anonymous=True).status_code == 401 + + +def test_create_user_can_be_admin(gitea: GiteaInstance, unique_name: str) -> None: + login, password = gitea.create_user(unique_name, admin=True) + token = gitea.mint_token(login, password, ["read:user"]) + assert gitea.api("GET", "/user", token=token).json()["is_admin"] is True + + +def test_create_org_and_repos(gitea: GiteaInstance, unique_name: str) -> None: + org = gitea.create_org(unique_name) + assert org["name"] == unique_name + + org_repo = gitea.create_repo(unique_name, "org-repo", org=True) + assert org_repo["full_name"] == f"{unique_name}/org-repo" + assert org_repo["empty"] is False + + user_repo = gitea.create_repo(gitea.admin_login, f"{unique_name}-u", auto_init=False) + assert user_repo["owner"]["login"] == gitea.admin_login + assert user_repo["empty"] is True + + fetched = gitea.api("GET", f"/repos/{unique_name}/org-repo") + assert fetched.status_code == 200 + + +def test_unique_names_differ(gitea: GiteaInstance) -> None: + assert gitea.unique("x") != gitea.unique("x") + assert gitea.unique("x").startswith("x-") + + +def test_secrets_never_in_repr_or_errors( + gitea: GiteaInstance, + regular_user: RegularUser, + capfd: pytest.CaptureFixture[str], +) -> None: + secrets_ = [ + gitea.admin_token, + gitea.admin_password, + regular_user.token, + regular_user.password, + ] + rendered = repr(gitea) + repr(regular_user) + str(gitea) + str(regular_user) + with pytest.raises(HarnessError) as excinfo: + gitea.mint_token(regular_user.login, "definitely-wrong-password", ["all"]) + gitea.create_org(gitea.unique("quiet")) + out, err = capfd.readouterr() + for secret in secrets_: + assert secret not in rendered + assert secret not in str(excinfo.value) + assert secret not in out + assert secret not in err + + +def test_cleanup_removes_container_and_network() -> None: + run_id = harness.secrets.token_hex(4) + name = f"{harness.NAME_PREFIX}{run_id}" + res = RunResources(run_id=run_id, container=name, network=name) + harness.register_resources(res) + harness.run_docker("network", "create", name) + harness.run_docker("create", "--name", name, "--network", name, harness.GITEA_IMAGE) + assert _names("container", run_id) == name + assert _names("network", run_id) == name + + harness.cleanup_resources(res) + + assert _names("container", run_id) == "" + assert _names("network", run_id) == "" + # Idempotent, and no longer tracked for atexit. + harness.cleanup_resources(res) + assert run_id not in harness._ACTIVE + + +def test_cleanup_all_handles_registered_runs() -> None: + run_id = harness.secrets.token_hex(4) + name = f"{harness.NAME_PREFIX}{run_id}" + harness.register_resources(RunResources(run_id=run_id, container=name, network=name)) + harness.run_docker("network", "create", name) + harness.cleanup_all() + assert _names("network", run_id) == "" + + +def test_missing_docker_cli_gives_clear_error(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("PATH", "") + with pytest.raises(HarnessError, match="Docker is unavailable"): + harness.require_docker()