Spark
d51ef5a970
CI / test (push) Successful in 1m53s
- Comptes en base (PBKDF2), rôles, désactivation qui tue la session, audit ; secours environnement pour l'opérateur ; liaison draco par utilisateur ; page /profile. - Espaces de travail : connexions, membres, sélection de projets ; board/notes/épingles/journal scopés par espace ; sélecteur d'espace, badge connexion. - Panneaux /admin (utilisateurs, connexions, espaces, paramètres — l'environnement amorce, le panneau gouverne). - /archives : archive locale réversible, archivage forge confirmé, purge de l'état local, export Markdown, file « à archiver ? » avec archivage en lot. - Correctif session : les comptes membres résolus par la base (la session n'est plus limitée à l'opérateur d'environnement). - 231 tests verts (12 nouveaux au palier 19).
361 lines
12 KiB
Python
361 lines
12 KiB
Python
"""Palier 19 (0.24.0): multi-forges, utilisateurs, espaces,
|
|
paramètres en base, archives — P0 à P4 de l'analyse du 2026-10-21."""
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app import accounts, forge_crypto, models
|
|
from tests.conftest import csrf_from
|
|
|
|
|
|
# ── Migration des anciennes bases (P0) ────────────────────
|
|
|
|
def test_migration_legacy_schema_gains_scope(tmp_path, monkeypatch):
|
|
db = tmp_path / "legacy.db"
|
|
conn = sqlite3.connect(db)
|
|
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',
|
|
position REAL NOT NULL DEFAULT 0,
|
|
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')),
|
|
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
UNIQUE(repo_full_name)
|
|
);
|
|
CREATE TABLE today_picks (
|
|
pick_date TEXT NOT NULL,
|
|
repo_full_name TEXT NOT NULL,
|
|
issue_number INTEGER NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
PRIMARY KEY (pick_date, repo_full_name, issue_number)
|
|
);
|
|
CREATE TABLE board_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
repo_full_name TEXT NOT NULL,
|
|
issue_number INTEGER NOT NULL,
|
|
event_type TEXT NOT NULL,
|
|
from_column TEXT,
|
|
to_column TEXT,
|
|
source TEXT NOT NULL DEFAULT 'ui',
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
INSERT INTO board_state (repo_full_name, issue_number, column_name)
|
|
VALUES ('Projets/demo', 1, 'todo');
|
|
INSERT INTO project_notes (repo_full_name, note)
|
|
VALUES ('Projets/demo', 'note héritée');
|
|
"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
monkeypatch.setenv("KANBANTE_DB_PATH", str(db))
|
|
models.init_db()
|
|
board = models.get_board_state("Projets/demo")
|
|
assert [c["issue_number"] for c in board["todo"]] == [1]
|
|
assert models.get_notes("Projets/demo") == "note héritée"
|
|
# La portée par défaut est (1, 1) et les nouvelles tables existent.
|
|
assert models.list_connections() == []
|
|
assert models.count_users() == 0
|
|
|
|
|
|
# ── Chiffrement des jetons (P0) ───────────────────────────
|
|
|
|
def test_forge_token_encryption_roundtrip(monkeypatch):
|
|
from cryptography.fernet import Fernet
|
|
|
|
monkeypatch.setenv("KANBANTE_MASTER_KEY", Fernet.generate_key().decode())
|
|
forge_crypto._fernet = None # forcer la relecture de la clé
|
|
stored = forge_crypto.encrypt_token("ghp_secret")
|
|
assert stored != "ghp_secret"
|
|
assert forge_crypto.decrypt_token(stored) == "ghp_secret"
|
|
assert forge_crypto.mask_token(stored) == "••••••••"
|
|
assert forge_crypto.encrypt_token("") == ""
|
|
|
|
|
|
# ── Amorçage (P0/P2/P3) ───────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def auth_client(tmp_path, monkeypatch, fake_gitea):
|
|
monkeypatch.setenv("KANBANTE_DB_PATH", str(tmp_path / "auth.db"))
|
|
monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno")
|
|
monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse")
|
|
with TestClient(main.app) as client:
|
|
yield client
|
|
|
|
|
|
def _login(client, username="bruno", password="motdepasse"):
|
|
client.get("/login")
|
|
token = client.cookies.get("kanbante_csrf") or ""
|
|
return client.post(
|
|
"/login",
|
|
data={
|
|
"username": username,
|
|
"password": password,
|
|
"next": "/",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
|
|
def test_bootstrap_creates_legacy_connection_admin_workspace(auth_client):
|
|
connections = models.list_connections()
|
|
assert len(connections) == 1
|
|
assert connections[0]["slug"] == "gitea"
|
|
assert connections[0]["kind"] == "gitea"
|
|
workspaces = models.list_workspaces()
|
|
assert [ws["slug"] for ws in workspaces] == ["principal"]
|
|
assert models.workspace_connection_ids(workspaces[0]["id"]) == [
|
|
connections[0]["id"]
|
|
]
|
|
users = models.list_users()
|
|
assert len(users) == 1
|
|
assert users[0]["username"] == "bruno"
|
|
assert users[0]["role"] == "admin"
|
|
assert accounts.verify_password("motdepasse", users[0]["password_hash"])
|
|
|
|
|
|
def test_db_login_and_password_change(auth_client):
|
|
# Le mot de passe d'environnement a créé le hash en base :
|
|
# la connexion passe par la base.
|
|
r = _login(auth_client)
|
|
assert r.status_code == 303
|
|
token = csrf_from(auth_client)
|
|
r = auth_client.post(
|
|
"/profile/password",
|
|
data={
|
|
"current_password": "motdepasse",
|
|
"new_password": "nouveau-secret",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
auth_client.get("/logout")
|
|
# Le nouveau mot de passe fonctionne…
|
|
assert _login(auth_client, password="nouveau-secret").status_code == 303
|
|
auth_client.get("/logout")
|
|
# …et pour un compte membre, l'ancien ne passe plus (pas de
|
|
# secours environnement pour les membres).
|
|
token = csrf_from(auth_client)
|
|
_login(auth_client, password="nouveau-secret")
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/admin/users",
|
|
data={
|
|
"username": "alice",
|
|
"password": "alice-pass",
|
|
"role": "member",
|
|
"display_name": "Alice",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
auth_client.get("/logout")
|
|
assert _login(auth_client, "alice", "alice-pass").status_code == 303
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/profile/password",
|
|
data={
|
|
"current_password": "alice-pass",
|
|
"new_password": "alice-nouveau",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
auth_client.get("/logout")
|
|
r = _login(auth_client, "alice", "alice-pass")
|
|
assert r.status_code == 401
|
|
assert _login(auth_client, "alice", "alice-nouveau").status_code == 303
|
|
|
|
|
|
def test_member_cannot_open_admin(auth_client):
|
|
_login(auth_client)
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/admin/users",
|
|
data={
|
|
"username": "bob",
|
|
"password": "bob-pass",
|
|
"role": "member",
|
|
"display_name": "",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
auth_client.get("/logout")
|
|
_login(auth_client, "bob", "bob-pass")
|
|
assert auth_client.get("/admin").status_code == 403
|
|
assert auth_client.get("/admin/connections").status_code == 403
|
|
assert auth_client.get("/profile").status_code == 200
|
|
|
|
|
|
def test_disabled_user_session_dies(auth_client):
|
|
_login(auth_client)
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/admin/users",
|
|
data={
|
|
"username": "eve",
|
|
"password": "eve-pass",
|
|
"role": "member",
|
|
"display_name": "",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
eve_id = models.get_user_by_username("eve")["id"]
|
|
auth_client.get("/logout")
|
|
_login(auth_client, "eve", "eve-pass")
|
|
assert auth_client.get("/profile").status_code == 200
|
|
# L'admin désactive Eve pendant que sa session est ouverte.
|
|
with TestClient(main.app) as admin_client:
|
|
_login(admin_client)
|
|
token = csrf_from(admin_client)
|
|
admin_client.post(
|
|
f"/admin/users/{eve_id}/toggle",
|
|
data={"csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
r = auth_client.get(
|
|
"/profile", headers={"Accept": "text/html"}, follow_redirects=False
|
|
)
|
|
assert r.status_code == 303
|
|
assert r.headers["location"].startswith("/login")
|
|
|
|
|
|
def test_admin_creates_github_connection_encrypted(auth_client):
|
|
_login(auth_client)
|
|
token = csrf_from(auth_client)
|
|
r = auth_client.post(
|
|
"/admin/connections",
|
|
data={
|
|
"kind": "github",
|
|
"name": "GitHub perso",
|
|
"base_url": "https://api.github.com",
|
|
"token": "ghp_secret_token",
|
|
"hook_secret": "",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
row = models.get_connection_by_slug("github-perso")
|
|
assert row is not None
|
|
assert row["kind"] == "github"
|
|
assert row["token_encrypted"] != "ghp_secret_token"
|
|
assert forge_crypto.decrypt_token(row["token_encrypted"]) == "ghp_secret_token"
|
|
page = auth_client.get("/admin/connections")
|
|
assert "ghp_secret_token" not in page.text # jamais réaffiché
|
|
assert "••••••••" in page.text
|
|
|
|
|
|
# ── Paramètres en base (P3) ───────────────────────────────
|
|
|
|
def test_settings_panel_overrides_env(auth_client):
|
|
from app.config import get_settings
|
|
|
|
_login(auth_client)
|
|
assert get_settings().aging_days == 7
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/admin/settings",
|
|
data={"setting_aging_days": "42", "csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert get_settings().aging_days == 42
|
|
# Champ vidé = retour à l'environnement.
|
|
token = csrf_from(auth_client)
|
|
auth_client.post(
|
|
"/admin/settings",
|
|
data={"setting_aging_days": "", "csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert get_settings().aging_days == 7
|
|
|
|
|
|
# ── Espaces et portées (P3) ───────────────────────────────
|
|
|
|
def test_board_state_is_scoped_per_workspace():
|
|
models.move_issue("Projets/demo", 1, "todo", workspace_id=1, connection_id=1)
|
|
models.move_issue("Projets/demo", 1, "review", workspace_id=2, connection_id=1)
|
|
assert models.get_board_state("Projets/demo", workspace_id=1)["todo"]
|
|
board2 = models.get_board_state("Projets/demo", workspace_id=2)
|
|
assert board2["review"] and not board2["todo"]
|
|
|
|
|
|
def test_workspace_switch_scopes_dashboard(client, fake_gitea):
|
|
# Un espace vide (aucune connexion) ne montre aucun projet.
|
|
empty_id = models.create_workspace("vide", "Vide", "#999999")
|
|
client.cookies.set("kanbante_workspace", str(empty_id))
|
|
page = client.get("/")
|
|
assert "/board/Projets/demo" not in page.text
|
|
principal = models.list_workspaces()[0]
|
|
client.cookies.set("kanbante_workspace", str(principal["id"]))
|
|
page = client.get("/")
|
|
assert "/board/Projets/demo" in page.text
|
|
|
|
|
|
# ── Archives (P4) ─────────────────────────────────────────
|
|
|
|
def test_local_archive_hides_and_restores(client, fake_gitea):
|
|
token = csrf_from(client)
|
|
principal = models.list_workspaces()[0]
|
|
assert "/board/Projets/demo" in client.get("/").text
|
|
r = client.post(
|
|
"/archives/local",
|
|
data={
|
|
"conn_id": "1",
|
|
"repo_full_name": "Projets/demo",
|
|
"archived": "1",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
assert "/board/Projets/demo" not in client.get("/").text
|
|
archives = client.get("/archives")
|
|
assert "Projets/demo" in archives.text
|
|
assert models.list_local_archived(principal["id"])
|
|
token = csrf_from(client)
|
|
client.post(
|
|
"/archives/local",
|
|
data={
|
|
"conn_id": "1",
|
|
"repo_full_name": "Projets/demo",
|
|
"archived": "0",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert "/board/Projets/demo" in client.get("/").text
|
|
|
|
|
|
def test_forge_archive_via_page(client, fake_gitea):
|
|
token = csrf_from(client)
|
|
r = client.post(
|
|
"/archives/forge",
|
|
data={
|
|
"conn_id": "1",
|
|
"repo_full_name": "Projets/demo",
|
|
"archived": "1",
|
|
"csrf_token": token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
repo = next(r for r in fake_gitea.repos if r["full_name"] == "Projets/demo")
|
|
assert repo["archived"] is True
|