"""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 ""