test: suite pytest (36 tests) et CI Gitea Actions

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

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

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

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

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

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

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

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