crafting-table/tests/conftest.py

"""Shared pytest fixtures for crafting-table tests.

Every test gets:
- A fresh tmp dir for db / log_dir / workspace_root / admin_bearer file
- Module reload so server.py's module-level singletons pick up our env
- IP allowlist patched to accept 127.0.0.1 (TestClient reports 'testclient')
- An admin bearer + a pre-minted app token + a second app token for isolation tests

Tests can then use `client` (TestClient + ctx dict) for HTTP-level tests, or
`db_only` if they only need a DB instance (no server boot).
"""
from __future__ import annotations

import os
import sys
import time
from pathlib import Path

import pytest


# ---------- per-test workspace --------------------------------------------


@pytest.fixture
def tmp_workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    db_path = tmp_path / "crafting.db"
    log_dir = tmp_path / "jobs"
    workspace_root = tmp_path / "workspace"
    admin_bearer = tmp_path / "admin-bearer.txt"
    log_dir.mkdir()
    workspace_root.mkdir()

    monkeypatch.setenv("CRAFTING_DB", str(db_path))
    monkeypatch.setenv("CRAFTING_LOG_DIR", str(log_dir))
    monkeypatch.setenv("CRAFTING_WORKSPACE", str(workspace_root))
    monkeypatch.setenv("CRAFTING_ADMIN_BEARER", str(admin_bearer))
    monkeypatch.setenv("CRAFTING_MAX_CONCURRENT", "2")
    monkeypatch.setenv("CRAFTING_DEFAULT_JOB_TIMEOUT", "10")
    monkeypatch.setenv("CRAFTING_LAN_CIDRS", "127.0.0.0/8,::1/128,10.0.0.0/8")
    monkeypatch.setenv("CRAFTING_GC_INTERVAL", "9999")
    monkeypatch.setenv("CRAFTING_GC_AGE", "86400")

    yield {
        "db_path": db_path,
        "log_dir": log_dir,
        "workspace_root": workspace_root,
        "admin_bearer": admin_bearer,
    }


@pytest.fixture
def db_only(tmp_workspace):
    """For tests that don't need the server — just a DB instance."""
    # Import locally so env vars are in effect before module import.
    from crafting_table.db import DB
    return DB(str(tmp_workspace["db_path"]))


# ---------- server reload + TestClient ------------------------------------


def _reload_server_modules():
    """Drop cached modules so module-level singletons re-bind with current env."""
    for name in [
        "crafting_table.server",
        "crafting_table.runner",
        "crafting_table.workspace",
        "crafting_table.auth",
        "crafting_table.db",
        "crafting_table.config",
        "crafting_table.models",
        "crafting_table",
    ]:
        sys.modules.pop(name, None)


@pytest.fixture
def server(tmp_workspace, monkeypatch: pytest.MonkeyPatch):
    """Reload server module + return the module so tests can poke internals."""
    _reload_server_modules()

    from crafting_table import auth as auth_mod  # noqa: WPS433
    monkeypatch.setattr(auth_mod, "_client_ip", lambda _req: "127.0.0.1")

    from crafting_table import server as srv  # noqa: WPS433
    return srv


@pytest.fixture
def client(server, tmp_workspace):
    """FastAPI TestClient + ctx dict (admin_bearer, app_token, other_token, ...).

    The TestClient context-manages lifespan, so on entry the runner starts +
    admin token is bootstrapped. On exit the runner stops cleanly.
    """
    from fastapi.testclient import TestClient

    with TestClient(server.app) as tc:
        admin_bearer = tmp_workspace["admin_bearer"].read_text().strip()

        # Mint two app tokens for isolation tests
        r = tc.post(
            "/admin/tokens",
            headers={"Authorization": f"Bearer {admin_bearer}"},
            json={"name": "alpha", "is_admin": False, "ip_cidrs": []},
        )
        assert r.status_code == 200, r.text
        alpha_bearer = r.json()["bearer"]

        r2 = tc.post(
            "/admin/tokens",
            headers={"Authorization": f"Bearer {admin_bearer}"},
            json={"name": "bravo", "is_admin": False, "ip_cidrs": []},
        )
        assert r2.status_code == 200, r2.text
        bravo_bearer = r2.json()["bearer"]

        yield tc, {
            "admin_bearer": admin_bearer,
            "alpha_bearer": alpha_bearer,
            "alpha_name": "alpha",
            "bravo_bearer": bravo_bearer,
            "bravo_name": "bravo",
            "server": server,
        }


# ---------- shared sample data ---------------------------------------------


def sample_project_payload(*, name: str = "demo", recipe_cmds: dict | None = None) -> dict:
    cmds = recipe_cmds or {
        "build": "echo build && true",
        "test": "echo test && true",
        "lint": "echo lint && true",
        "audit": "echo audit && true",
    }
    return {
        "name": name,
        "git_url": "/dev/null",  # tests don't actually clone unless they want to
        "default_branch": "main",
        "languages": ["python"],
        "subprojects": [
            {
                "path": ".",
                "language": "python",
                "build": cmds.get("build"),
                "test": cmds.get("test"),
                "lint": cmds.get("lint"),
                "audit": cmds.get("audit"),
                "timeout_secs": 5,
            }
        ],
        "schedule": {"audit": "manual", "test": None, "build": None, "lint": None},
        "notify": {"email": [], "on": [], "auto_patch": False},
        "created_at": 0,
        "updated_at": 0,
    }