Spark
b99e87ccca
FakeGitea: get_user, search_issues (filtres q/état/assigné, PR exclues, repository joint), delete_label, forget_label_cache; issue 1 assignée à bruno pour /my.
278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""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": [{"id": 900, "name": "bug", "color": "ff0000"}] if number == 1 else [],
|
|
"assignee": {"login": "bruno", "avatar_url": ""} if number == 1 else None,
|
|
"assignees": [{"login": "bruno", "avatar_url": ""}] if number == 1 else [],
|
|
"milestone": {"id": 1, "title": "Sprint 1"} if number in (1, 3) else 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.repo_labels: dict[str, dict[str, dict]] = {"Projets/demo": {}}
|
|
self.label_ops: list[tuple] = []
|
|
self._label_seq = 100
|
|
self.cache_invalidations = 0
|
|
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 get_user(self):
|
|
return {"id": 1, "login": "bruno"}
|
|
|
|
async def search_issues(self, q="", state="open", assigned_by=None):
|
|
results = []
|
|
for full_name, issues in self.issues.items():
|
|
for issue in issues:
|
|
if issue.get("pull_request"):
|
|
continue
|
|
if state != "all" and issue["state"] != state:
|
|
continue
|
|
if q and q.lower() not in issue["title"].lower():
|
|
continue
|
|
if assigned_by:
|
|
logins = [a["login"] for a in issue.get("assignees") or []]
|
|
if issue.get("assignee"):
|
|
logins.append(issue["assignee"]["login"])
|
|
if assigned_by not in logins:
|
|
continue
|
|
results.append({**issue, "repository": {"full_name": full_name}})
|
|
return results
|
|
|
|
async def delete_label(self, owner, repo, label_id):
|
|
labels = self.repo_labels.setdefault(f"{owner}/{repo}", {})
|
|
for name in [n for n, l in labels.items() if l["id"] == label_id]:
|
|
del labels[name]
|
|
self.label_ops.append(("delete", f"{owner}/{repo}", label_id))
|
|
|
|
def forget_label_cache(self, owner, repo):
|
|
pass
|
|
|
|
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
|
|
|
|
def invalidate_cache(self):
|
|
self.cache_invalidations += 1
|
|
|
|
async def get_labels(self, owner, repo):
|
|
return list(self.repo_labels.setdefault(f"{owner}/{repo}", {}).values())
|
|
|
|
async def create_label(self, owner, repo, name, color):
|
|
self._label_seq += 1
|
|
label = {"id": self._label_seq, "name": name, "color": color.lstrip("#")}
|
|
self.repo_labels.setdefault(f"{owner}/{repo}", {})[name] = label
|
|
self.label_ops.append(("create", f"{owner}/{repo}", name))
|
|
return label
|
|
|
|
async def ensure_label(self, owner, repo, name, color):
|
|
existing = self.repo_labels.setdefault(f"{owner}/{repo}", {})
|
|
if name in existing:
|
|
return existing[name]
|
|
return await self.create_label(owner, repo, name, color)
|
|
|
|
async def add_issue_label(self, owner, repo, number, label_id):
|
|
issue = await self.get_issue(owner, repo, number)
|
|
label = next(
|
|
l for l in self.repo_labels[f"{owner}/{repo}"].values()
|
|
if l["id"] == label_id
|
|
)
|
|
if label not in issue["labels"]:
|
|
issue["labels"].append(label)
|
|
self.label_ops.append(("add", number, label["name"]))
|
|
|
|
async def remove_issue_label(self, owner, repo, number, label_id):
|
|
issue = await self.get_issue(owner, repo, number)
|
|
removed = [l for l in issue["labels"] if l["id"] == label_id]
|
|
issue["labels"] = [l for l in issue["labels"] if l["id"] != label_id]
|
|
for label in removed:
|
|
self.label_ops.append(("remove", number, label["name"]))
|
|
|
|
async def get_milestones(self, owner, repo, state="open"):
|
|
return [
|
|
{
|
|
"id": 1, "title": "Sprint 1",
|
|
"due_on": "2026-10-20T00:00:00Z",
|
|
"open_issues": 1, "closed_issues": 1,
|
|
}
|
|
]
|
|
|
|
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 ""
|