From 5aa0c7d9a70b20940b0b2243fd68952c61bf507a Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 09:31:48 -0400 Subject: [PATCH] test: suite pytest (36 tests) et CI Gitea Actions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Modèles (chemin DB, migration 0.1.0, ordre, réconciliation), classification, client Gitea (pagination, en-tête, écritures), API (CSRF, write-back, codes d'erreur, 502 propre), auth Basic et fail-closed. Tous verts; pyflakes propre. --- .gitea/workflows/ci.yml | 20 ++++ tests/conftest.py | 151 ++++++++++++++++++++++++++++ tests/test_api.py | 198 +++++++++++++++++++++++++++++++++++++ tests/test_auth.py | 36 +++++++ tests/test_classify.py | 44 +++++++++ tests/test_gitea_client.py | 98 ++++++++++++++++++ tests/test_models.py | 99 +++++++++++++++++++ 7 files changed, 646 insertions(+) create mode 100644 .gitea/workflows/ci.yml create mode 100644 tests/conftest.py create mode 100644 tests/test_api.py create mode 100644 tests/test_auth.py create mode 100644 tests/test_classify.py create mode 100644 tests/test_gitea_client.py create mode 100644 tests/test_models.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..0ab9b8e --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,20 @@ +name: CI + +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install + run: pip install -e ".[dev]" + - name: Compile check + run: python -m py_compile app/*.py + - name: Tests + run: pytest -q diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d0018a9 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,151 @@ +"""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 "" diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..e8898e1 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,198 @@ +"""HTTP API behaviour with a fake Gitea (audit C2/C3/I5/M6/M7).""" + +from app import models +from tests.conftest import csrf_from + + +def test_health_is_public_liveness(client): + r = client.get("/api/health") + assert r.status_code == 200 + assert r.json()["status"] == "ok" + assert r.json()["version"] == "0.2.0" + + +def test_readiness_reflects_gitea(client, fake_gitea): + assert client.get("/api/readiness").status_code == 200 + fake_gitea.ping_result = False + r = client.get("/api/readiness") + assert r.status_code == 503 + assert r.json()["checks"]["gitea"] is False + assert r.json()["checks"]["database"] is True + + +def test_dashboard_renders_three_categories(client): + r = client.get("/") + assert r.status_code == 200 + assert "demo" in r.text and "vieux" in r.text and "archive" in r.text + assert "Archivés" in r.text and "Inactifs" in r.text + + +def test_board_hides_pull_requests_and_reconciles(client): + models.move_issue("Projets/demo", 3, "todo") # closed in Gitea + r = client.get("/board/Projets/demo") + assert r.status_code == 200 + assert "Tâche un" in r.text + assert "PR déguisée" not in r.text # M6: PRs are not issues + state = models.get_board_state("Projets/demo") + assert [c["issue_number"] for c in state["done"]] == [3] # reconciled + + +def test_focus_lists_in_progress_across_projects(client): + models.move_issue("Projets/demo", 5, "in_progress") + r = client.get("/focus") + assert r.status_code == 200 + assert "Tâche focus" in r.text + + +def test_gitea_outage_is_a_clean_502_not_500(client, fake_gitea): + fake_gitea.raise_on_search = True + r = client.get("/") + assert r.status_code == 502 + assert "joindre Gitea" in r.text + r = client.get("/api/projects") + assert r.status_code == 502 + assert r.json()["ok"] is False + + +# ── CSRF + moves + write-back ────────────────────────────── + +def test_move_requires_csrf_token(client): + r = client.post( + "/api/move", + data={"repo_full_name": "Projets/demo", "issue_number": 1, "to_column": "done"}, + ) + assert r.status_code == 403 + + +def test_move_to_done_closes_issue_in_gitea(client, fake_gitea): + token = csrf_from(client) + r = client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", + "issue_number": 1, + "to_column": "done", + "from_column": "backlog", + "to_order": "1", + "csrf_token": token, + }, + ) + assert r.status_code == 200 + assert r.json()["gitea_sync"] == "closed" + assert fake_gitea.state_changes == [("Projets", "demo", 1, "closed")] + state = models.get_board_state("Projets/demo") + assert [c["issue_number"] for c in state["done"]] == [1] + + +def test_move_out_of_done_reopens_issue(client, fake_gitea): + token = csrf_from(client) + client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", "issue_number": 1, + "to_column": "done", "from_column": "backlog", "csrf_token": token, + }, + ) + r = client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", "issue_number": 1, + "to_column": "todo", "from_column": "done", "csrf_token": token, + }, + ) + assert r.json()["gitea_sync"] == "reopened" + assert fake_gitea.state_changes[-1] == ("Projets", "demo", 1, "open") + + +def test_move_writeback_failure_is_reported_not_fatal(client, fake_gitea): + fake_gitea.fail_state_change = True + token = csrf_from(client) + r = client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", "issue_number": 1, + "to_column": "done", "from_column": "backlog", "csrf_token": token, + }, + ) + assert r.status_code == 200 + assert r.json()["ok"] is True + assert r.json()["gitea_sync"] == "failed" + + +def test_move_invalid_column_is_400(client): + token = csrf_from(client) + r = client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", "issue_number": 1, + "to_column": "nimporte_quoi", "csrf_token": token, + }, + ) + assert r.status_code == 400 + assert r.json()["ok"] is False + + +def test_move_persists_card_order(client): + token = csrf_from(client) + r = client.post( + "/api/move", + data={ + "repo_full_name": "Projets/demo", "issue_number": 5, + "to_column": "backlog", "from_column": "backlog", + "to_order": "5,1", "csrf_token": token, + }, + ) + assert r.status_code == 200 + state = models.get_board_state("Projets/demo") + assert [c["issue_number"] for c in state["backlog"]][:2] == [5, 1] + + +# ── Notes + creation ─────────────────────────────────────── + +def test_notes_roundtrip_via_api(client): + token = csrf_from(client) + r = client.post( + "/api/notes", + data={"repo_full_name": "Projets/demo", "note": "à retenir", "csrf_token": token}, + ) + assert r.status_code == 200 + assert models.get_notes("Projets/demo") == "à retenir" + + +def test_notes_size_limit_is_413(client, monkeypatch): + monkeypatch.setenv("KANBANTE_MAX_NOTE_CHARS", "10") + token = csrf_from(client) + r = client.post( + "/api/notes", + data={ + "repo_full_name": "Projets/demo", + "note": "x" * 50, + "csrf_token": token, + }, + ) + assert r.status_code == 413 + + +def test_create_issue_goes_to_gitea_and_lands_on_board(client, fake_gitea): + token = csrf_from(client) + r = client.post( + "/api/issues", + data={ + "repo_full_name": "Projets/demo", "title": "Bug à corriger", + "column": "backlog", "csrf_token": token, + }, + ) + assert r.status_code == 201 + number = r.json()["issue"]["number"] + assert fake_gitea.created == [("Projets", "demo", "Bug à corriger")] + state = models.get_board_state("Projets/demo") + assert number in [c["issue_number"] for c in state["backlog"]] + + +def test_create_issue_validates_title(client): + token = csrf_from(client) + r = client.post( + "/api/issues", + data={"repo_full_name": "Projets/demo", "title": " ", "csrf_token": token}, + ) + assert r.status_code == 400 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..bdcb13b --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,36 @@ +"""Basic auth + fail-closed behaviour (audit C2).""" + +import base64 + +from fastapi.testclient import TestClient + +import app.main as main + + +def _basic(user: str, password: str) -> dict: + raw = base64.b64encode(f"{user}:{password}".encode()).decode() + return {"Authorization": f"Basic {raw}"} + + +def test_no_credentials_means_open_but_health_works(client): + assert client.get("/").status_code == 200 + + +def test_auth_required_when_configured(monkeypatch, fake_gitea): + monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno") + monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse") + with TestClient(main.app) as client: + assert client.get("/").status_code == 401 + assert client.get("/", headers=_basic("bruno", "mauvais")).status_code == 401 + assert client.get("/", headers=_basic("bruno", "motdepasse")).status_code == 200 + # Liveness stays public for Docker HEALTHCHECK. + assert client.get("/api/health").status_code == 200 + + +def test_require_auth_fails_closed_without_credentials(monkeypatch, fake_gitea): + monkeypatch.delenv("KANBANTE_AUTH_USER", raising=False) + monkeypatch.delenv("KANBANTE_AUTH_PASSWORD", raising=False) + monkeypatch.setenv("KANBANTE_REQUIRE_AUTH", "1") + with TestClient(main.app) as client: + assert client.get("/").status_code == 503 + assert client.get("/api/health").status_code == 200 diff --git a/tests/test_classify.py b/tests/test_classify.py new file mode 100644 index 0000000..3b11054 --- /dev/null +++ b/tests/test_classify.py @@ -0,0 +1,44 @@ +"""Repo classification: rolling window + real archived flag (audit I1).""" + +from datetime import datetime, timedelta, timezone + +from app.main import classify_repos + +NOW = datetime(2026, 10, 11, tzinfo=timezone.utc) + + +def _repo(name, updated, archived=False): + return { + "full_name": f"Projets/{name}", + "updated_at": updated.strftime("%Y-%m-%dT%H:%M:%SZ"), + "archived": archived, + } + + +def test_recent_repo_is_active(): + active, inactive, archived = classify_repos( + [_repo("a", NOW - timedelta(days=2))], now=NOW + ) + assert [r["full_name"] for r in active] == ["Projets/a"] + assert inactive == [] and archived == [] + + +def test_old_repo_is_inactive_not_archived(): + active, inactive, archived = classify_repos( + [_repo("a", NOW - timedelta(days=900))], now=NOW + ) + assert inactive and not active and not archived + + +def test_archived_flag_wins_even_if_recent(): + # 0.1.0 showed recently-touched archived repos as "active". + active, inactive, archived = classify_repos( + [_repo("a", NOW - timedelta(hours=1), archived=True)], now=NOW + ) + assert archived and not active and not inactive + + +def test_no_hardcoded_year_2027_still_works(): + future = datetime(2027, 1, 5, tzinfo=timezone.utc) + active, _, _ = classify_repos([_repo("a", future - timedelta(days=3))], now=future) + assert len(active) == 1 diff --git a/tests/test_gitea_client.py b/tests/test_gitea_client.py new file mode 100644 index 0000000..b115b0a --- /dev/null +++ b/tests/test_gitea_client.py @@ -0,0 +1,98 @@ +"""Gitea client: pagination, auth header, write operations (audit I2/M9).""" + +import httpx + +from app.gitea_client import PAGE_SIZE, GiteaClient + + +def _client(handler, token=""): + return GiteaClient( + base_url="http://gitea.test", + token=token, + cache_ttl=0, + transport=httpx.MockTransport(handler), + ) + + +async def test_pagination_follows_until_short_page(): + seen_pages = [] + + def handler(request): + if request.url.path.endswith("/user"): + return httpx.Response(200, json={"id": 7}) + page = int(request.url.params.get("page", "1")) + seen_pages.append(page) + count = PAGE_SIZE if page == 1 else 25 + start = (page - 1) * PAGE_SIZE + data = [{"id": start + i} for i in range(count)] + return httpx.Response(200, json={"data": data}, headers={"X-Total-Count": "75"}) + + client = _client(handler) + repos = await client.search_repos() + assert len(repos) == 75 + assert seen_pages.count(1) >= 1 and 2 in seen_pages + + +async def test_pagination_without_total_header(): + def handler(request): + page = int(request.url.params.get("page", "1")) + count = PAGE_SIZE if page == 1 else 10 + return httpx.Response(200, json=[{"number": i} for i in range(count)]) + + client = _client(handler) + issues = await client.get_repo_issues("o", "r", state="open") + assert len(issues) == PAGE_SIZE + 10 + + +async def test_search_uses_authenticated_user_id_not_hardcoded(): + seen = {} + + def handler(request): + if request.url.path.endswith("/user"): + return httpx.Response(200, json={"id": 42}) + seen["uid"] = request.url.params.get("uid") + return httpx.Response(200, json={"data": []}) + + client = _client(handler) + await client.search_repos() + assert seen["uid"] == "42" + + +async def test_authorization_header_only_with_token(): + captured = {} + + def handler(request): + captured["auth"] = request.headers.get("authorization") + return httpx.Response(200, json={"version": "1.27.3"}) + + await _client(handler, token="").ping() + assert captured["auth"] is None + + await _client(handler, token="sekret").ping() + assert captured["auth"] == "token sekret" + + +async def test_set_issue_state_patches_gitea(): + captured = {} + + def handler(request): + captured["method"] = request.method + captured["path"] = request.url.path + captured["json"] = request.content.decode() + return httpx.Response(200, json={"state": "closed"}) + + client = _client(handler, token="x") + await client.set_issue_state("Projets", "demo", 3, "closed") + assert captured["method"] == "PATCH" + assert captured["path"].endswith("/repos/Projets/demo/issues/3") + assert '"closed"' in captured["json"] + + +async def test_create_issue_posts_title(): + def handler(request): + assert request.method == "POST" + return httpx.Response(201, json={"number": 9, "title": "Nouvelle"}) + + client = _client(handler, token="x") + issue = await client.create_issue("Projets", "demo", "Nouvelle") + assert issue["number"] == 9 diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..665df4b --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,99 @@ +"""Model layer: env-driven DB path, migrations, ordering, reconciliation.""" + +import os +import sqlite3 + +from app import models + + +def test_db_path_honours_env_var(): + # Regression for audit point C1: the path the Dockerfile sets must win. + assert models.db_path() == os.environ["KANBANTE_DB_PATH"] + + +def test_board_state_ignores_unknown_column_rows(): + # Regression for audit point I4: a legacy row must not KeyError the board. + with models.get_db() as conn: + conn.execute( + "INSERT INTO board_state (repo_full_name, issue_number, column_name) " + "VALUES ('a/b', 99, 'ancienne_colonne')" + ) + state = models.get_board_state("a/b") + assert all(cards == [] for cards in state.values()) + + +def test_migration_from_v01_schema(tmp_path, monkeypatch): + legacy = tmp_path / "legacy.db" + conn = sqlite3.connect(legacy) + conn.executescript( + """ + CREATE TABLE board_state ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repo_full_name TEXT NOT NULL, + issue_number INTEGER NOT NULL, + column_name TEXT NOT NULL DEFAULT 'backlog', + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + UNIQUE(repo_full_name, issue_number) + ); + CREATE TABLE project_notes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repo_full_name TEXT NOT NULL, + note TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + UNIQUE(repo_full_name) + ); + INSERT INTO board_state (repo_full_name, issue_number, column_name) + VALUES ('a/b', 1, 'todo'), ('a/b', 2, 'colonne_disparue'); + INSERT INTO project_notes (repo_full_name, note) VALUES ('a/b', 'salut'); + """ + ) + conn.commit() + conn.close() + + monkeypatch.setenv("KANBANTE_DB_PATH", str(legacy)) + models.init_db() + + with models.get_db() as conn: + board_cols = [r[1] for r in conn.execute("PRAGMA table_info(board_state)")] + notes_cols = [r[1] for r in conn.execute("PRAGMA table_info(project_notes)")] + remaining = conn.execute("SELECT COUNT(*) FROM board_state").fetchone()[0] + assert "position" in board_cols + assert "updated_at" in notes_cols + assert remaining == 1 # valid row kept, unknown-column row dropped + assert models.get_notes("a/b") == "salut" + + +def test_move_and_column_order_persisted(): + models.move_issue("a/b", 1, "todo") + models.move_issue("a/b", 2, "todo") + models.set_column_order("a/b", "todo", [2, 1]) + state = models.get_board_state("a/b") + assert [c["issue_number"] for c in state["todo"]] == [2, 1] + + +def test_reconcile_closed_to_done_and_purge_deleted(): + models.move_issue("a/b", 1, "in_progress") + models.move_issue("a/b", 2, "todo") + models.move_issue("a/b", 3, "review") + changed = models.reconcile_board_state("a/b", open_numbers={1}, closed_numbers={2}) + assert changed == {"closed_to_done": 1, "purged": 1} + state = models.get_board_state("a/b") + assert [c["issue_number"] for c in state["in_progress"]] == [1] + assert [c["issue_number"] for c in state["done"]] == [2] + assert all( + c["issue_number"] != 3 for cards in state.values() for c in cards + ) + + +def test_notes_roundtrip(): + assert models.get_notes("a/b") is None + models.set_notes("a/b", "première version") + models.set_notes("a/b", "deuxième version") + assert models.get_notes("a/b") == "deuxième version" + + +def test_move_issue_rejects_invalid_column(): + import pytest + + with pytest.raises(ValueError): + models.move_issue("a/b", 1, "pas_une_colonne")