kanbante/tests/conftest.py

"""Shared fixtures: isolated SQLite file + a fake async Gitea client.

Environment is set BEFORE any app module is imported so the app picks up
the test database path and never talks to a real Gitea instance.
"""

import os
import tempfile
from datetime import datetime, timedelta, timezone

_TMP = tempfile.mkdtemp(prefix="kanbante-tests-")
os.environ["KANBANTE_DB_PATH"] = os.path.join(_TMP, "test.db")
os.environ["GITEA_URL"] = "http://gitea.test"
os.environ.pop("GITEA_TOKEN", None)
os.environ.pop("KANBANTE_AUTH_USER", None)
os.environ.pop("KANBANTE_AUTH_PASSWORD", None)
os.environ.pop("KANBANTE_REQUIRE_AUTH", None)

import httpx  # noqa: E402
import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

import app.main as main  # noqa: E402
from app import models  # noqa: E402


def _iso(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _repo(name: str, updated: datetime, archived: bool = False) -> dict:
    return {
        "full_name": f"Projets/{name}",
        "name": name,
        "description": f"Repo {name}",
        "updated_at": _iso(updated),
        "open_issues_count": 1,
        "owner": {"login": "Projets"},
        "archived": archived,
        "html_url": f"http://gitea.test/Projets/{name}",
    }


def _issue(number: int, title: str, state: str = "open", pr: bool = False) -> dict:
    issue = {
        "number": number,
        "title": title,
        "state": state,
        "labels": [{"name": "bug", "color": "ff0000"}] if number == 1 else [],
        "assignee": None,
        "created_at": "2026-09-01T09:00:00Z",
        "html_url": f"http://gitea.test/Projets/demo/issues/{number}",
    }
    if pr:
        issue["pull_request"] = {"url": "http://gitea.test/pr"}
    return issue


class FakeGitea:
    """Drop-in async replacement for GiteaClient used by the API tests."""

    def __init__(self) -> None:
        now = datetime.now(timezone.utc)
        self.repos = [
            _repo("demo", now - timedelta(days=1)),
            _repo("vieux", now - timedelta(days=800)),
            _repo("archive", now - timedelta(days=1), archived=True),
        ]
        self.issues = {
            "Projets/demo": [
                _issue(1, "Tâche un"),
                _issue(2, "PR déguisée", pr=True),
                _issue(3, "Ancienne tâche", state="closed"),
                _issue(5, "Tâche focus"),
            ],
        }
        self.state_changes: list[tuple] = []
        self.created: list[tuple] = []
        self.ping_result = True
        self.raise_on_search = False
        self.fail_state_change = False
        self._next_number = 100

    async def aclose(self) -> None:
        pass

    async def ping(self) -> bool:
        return self.ping_result

    async def search_repos(self):
        if self.raise_on_search:
            raise httpx.HTTPError("simulated Gitea outage")
        return list(self.repos)

    async def get_repo(self, owner, repo):
        full = f"{owner}/{repo}"
        for r in self.repos:
            if r["full_name"] == full:
                return r
        return {"full_name": full, "html_url": "", "updated_at": ""}

    async def get_repo_issues(self, owner, repo, state="all"):
        items = list(self.issues.get(f"{owner}/{repo}", []))
        if state != "all":
            items = [i for i in items if i["state"] == state]
        return items

    async def create_issue(self, owner, repo, title, body=""):
        self._next_number += 1
        issue = _issue(self._next_number, title)
        issue["html_url"] = f"http://gitea.test/{owner}/{repo}/issues/{issue['number']}"
        self.issues.setdefault(f"{owner}/{repo}", []).append(issue)
        self.created.append((owner, repo, title))
        return issue

    async def set_issue_state(self, owner, repo, issue_number, state):
        if self.fail_state_change:
            raise httpx.HTTPError("simulated write-back failure")
        self.state_changes.append((owner, repo, issue_number, state))
        for issue in self.issues.get(f"{owner}/{repo}", []):
            if issue["number"] == issue_number:
                issue["state"] = "closed" if state == "closed" else "open"
        return {}


@pytest.fixture(autouse=True)
def clean_db():
    models.init_db()
    with models.get_db() as conn:
        conn.execute("DELETE FROM board_state")
        conn.execute("DELETE FROM project_notes")
    yield


@pytest.fixture
def fake_gitea(monkeypatch):
    fake = FakeGitea()
    monkeypatch.setattr(main, "gitea", fake)
    return fake


@pytest.fixture
def client(fake_gitea):
    with TestClient(main.app) as c:
        yield c


def csrf_from(client: TestClient) -> str:
    """Load a page so the app sets the CSRF cookie, then return its value."""
    client.get("/")
    return client.cookies.get("kanbante_csrf") or ""