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,
        "body": f"Corps de l'issue {number} en **markdown**.",
        "state": state,
        "labels": [{"name": "bug", "color": "ff0000"}] if number == 1 else [],
        "assignee": None,
        "assignees": [],
        "milestone": None,
        "due_date": None,
        "created_at": "2026-09-01T09:00:00Z",
        "updated_at": "2026-09-02T09: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.comments: dict[tuple, list[dict]] = {
            ("Projets/demo", 1): [
                {
                    "id": 1,
                    "body": "Premier commentaire",
                    "created_at": "2026-09-03T10:00:00Z",
                    "updated_at": "2026-09-03T10:00:00Z",
                    "html_url": "http://gitea.test/comment/1",
                    "user": {"login": "bruno", "avatar_url": ""},
                }
            ],
        }
        self.posted_comments: 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 get_issue(self, owner, repo, number):
        import httpx as _httpx

        for issue in self.issues.get(f"{owner}/{repo}", []):
            if issue["number"] == number:
                return issue
        request = _httpx.Request("GET", "http://gitea.test")
        raise _httpx.HTTPStatusError(
            "404", request=request, response=_httpx.Response(404, request=request)
        )

    async def get_issue_comments(self, owner, repo, number):
        return list(self.comments.get((f"{owner}/{repo}", number), []))

    async def create_comment(self, owner, repo, number, body):
        comment = {
            "id": 500 + len(self.posted_comments),
            "body": body,
            "created_at": "2026-10-11T10:00:00Z",
            "updated_at": "2026-10-11T10:00:00Z",
            "html_url": "http://gitea.test/comment/new",
            "user": {"login": "bruno", "avatar_url": ""},
        }
        self.comments.setdefault((f"{owner}/{repo}", number), []).append(comment)
        self.posted_comments.append((owner, repo, number, body))
        return comment

    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")
        conn.execute("DELETE FROM board_events")
    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 ""