"""FlowDeck — correctifs d'anomalies v7.46.0 (audit de fonctionnement). Chaque test verrouille une anomalie réellement constatée sur l'instance : * **P0-3** ``/auth/user`` renvoyait le ``password_hash`` (et le cookie de session, signé mais non chiffré, l'embarquait) ; * **P0-4** ``gitea_oauth_client_id`` vaut le placeholder ``test-id`` → ``/auth/login`` redirigeait vers Gitea avec un client_id invalide ; * **P0-1** écritures anonymes sur ``/api/workspaces*`` et ``/api/local-workspace/items`` (``uid = ... else 1``) ; * **P0-2** ``/workspace/*`` sans session : export CSV, historique, commentaires ; * **P1-5** bouton Home : ``local_workspaces[0]`` (ordre alphabétique) au lieu de l'espace actif ; * **P1-6** ``/workspace/favorites`` : 500 permanent (colonne ``collection_id`` supprimée par la migration v2.2.0) ; * **P1-7** ``/api/v2/agents/conversations`` masqué par ``/agents/{agent_id}`` ; * **P1-8** l'éditeur de page appelait ``/api/synced-blocks`` (404) au lieu de ``/board/api/synced-blocks`` ; * **P1-9** ``/board/api/synced-blocks`` : 500 anonyme + absence de contrôle de propriété sur PUT/DELETE. """ from __future__ import annotations import re import pytest from conftest import anon_csrf, login_test_client @pytest.fixture def client(): """Same isolated temp DB contract as tests/conftest.py::client.""" import os import tempfile db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) db_path = db_file.name db_file.close() backup_dir = tempfile.mkdtemp(prefix="fd_backups_") data_dir = tempfile.mkdtemp(prefix="fd_data_") os.environ["DATABASE_URL"] = f"sqlite:///{db_path}" os.environ["APP_SECRET_KEY"] = "test-secret-for-tests" os.environ["RATE_LIMIT_ENABLED"] = "false" os.environ["LLM_PROVIDER"] = "offline" os.environ["FLOWDECK_DATA_DIR"] = data_dir import app.config s = app.config.settings s.database_url = f"sqlite:///{db_path}" s.app_secret_key = "test-secret-for-tests" s.rate_limit_enabled = False s.backup_enabled = False s.backup_dir = backup_dir s.project_sync_enabled = False from app.db import init_db from app.main import app init_db() from fastapi.testclient import TestClient c = login_test_client(TestClient(app)) try: yield c finally: try: os.unlink(db_path) except FileNotFoundError: pass def _mk_ws(client, name: str, owner: int = 1) -> int: r = client.post("/api/workspaces", json={"name": name}) assert r.status_code == 200, r.text return r.json()["id"] # ══════════════════════ P0-3 — password_hash ══════════════════════ def test_auth_user_never_exposes_password_hash(client): r = client.get("/auth/user") assert r.status_code == 200 assert "password_hash" not in r.text def test_session_cookie_does_not_embed_password_hash(client): from app.auth.session import SessionManager, public_user sanitized = public_user({"id": 1, "login": "a", "password_hash": "deadbeef"}) assert "password_hash" not in sanitized raw = SessionManager.create_session( {"id": 1, "login": "a", "password_hash": "deadbeef", "is_admin": 1} ) assert "deadbeef" not in raw # Le payload décodé ne contient plus le champ sensible (décodé par la même # instance de serializer que le serveur, independamment de `settings`). decoded = SessionManager.decode_session(raw) assert decoded is not None assert "password_hash" not in decoded assert decoded["login"] == "a" # ══════════════════════ P0-4 — OAuth placeholder ══════════════════════ def test_placeholder_gitea_credentials_are_not_enabled(client): import app.auth.providers as providers gitea = providers.GiteaProvider( base_url="https://git.example.net", client_id="test-id", client_secret="test-secret", redirect_uri="", ) assert gitea.is_enabled() is False def test_login_does_not_redirect_to_placeholder_client(client): r = client.get("/auth/login", follow_redirects=False) assert r.status_code == 200 assert "client_id=test-id" not in r.text assert "not configured" in r.text.lower() def test_login_redirects_when_credentials_are_real(client): from app.config import settings old_id, old_secret = settings.gitea_oauth_client_id, settings.gitea_oauth_client_secret settings.gitea_oauth_client_id = "real-id" settings.gitea_oauth_client_secret = "real-secret" try: r = client.get("/auth/login", follow_redirects=False) assert r.status_code == 302 assert "client_id=real-id" in r.headers["location"] finally: settings.gitea_oauth_client_id = old_id settings.gitea_oauth_client_secret = old_secret # ══════════════════════ P0-1 — écritures anonymes ══════════════════════ @pytest.mark.parametrize( "method,path,body", [ ("post", "/api/workspaces", {"name": "ANON"}), ("put", "/api/workspaces/1", {"name": "ANON"}), ("delete", "/api/workspaces/1", None), ("post", "/api/workspaces/1/select", {}), ("post", "/api/local-workspace/items", {"name": "ANON"}), ("put", "/api/local-workspace/items/1", {"name": "ANON"}), ("delete", "/api/local-workspace/items/1", None), ("post", "/api/local-workspace/items/1/restore", {}), ("put", "/api/local-workspace/items/1/move", {"parent_id": None}), ], ) def test_anonymous_write_is_rejected(client, method, path, body): c = anon_csrf(client) fn = getattr(c, method) r = fn(path, json=body) if body is not None else fn(path) assert r.status_code == 401, f"{method.upper()} {path} -> {r.status_code} {r.text[:200]}" def test_anonymous_workspace_write_creates_nothing(client): ws_id = _mk_ws(client, "Real WS") c = anon_csrf(client) assert c.post("/api/workspaces", json={"name": "Ghost"}).status_code == 401 assert c.delete(f"/api/workspaces/{ws_id}").status_code == 401 from app.db import get_conn with get_conn() as conn: names = [r[0] for r in conn.execute("SELECT name FROM workspaces")] assert "Ghost" not in names assert "Real WS" in names def test_workspace_rename_delete_require_ownership(client): ws_id = _mk_ws(client, "WS of admin") other = login_test_client(client, user_id=2, login="intruder", is_admin=0) assert other.put(f"/api/workspaces/{ws_id}", json={"name": "hijacked"}).status_code == 403 assert other.delete(f"/api/workspaces/{ws_id}").status_code == 403 assert other.post(f"/api/workspaces/{ws_id}/select").status_code == 403 from app.db import get_conn with get_conn() as conn: assert conn.execute( "SELECT name FROM workspaces WHERE id=?", (ws_id,) ).fetchone()[0] == "WS of admin" # ══════════════════════ P0-2 — /workspace authentifié ══════════════════════ @pytest.mark.parametrize( "path", [ "/workspace/favorites", "/workspace/pages/1/comments", "/workspace/pages/1/history", "/workspace/templates/database", "/workspace/collections/1/export/csv", "/workspace/collections/1/sprints", "/workspace/collections/1/dashboards", ], ) def test_workspace_router_requires_session(client, path): c = anon_csrf(client) r = c.get(path) assert r.status_code == 401, f"{path} -> {r.status_code} {r.text[:160]}" def test_public_sharing_route_stays_public(client): """La seule route publique du router `/workspace` reste accessible.""" from app.db import get_conn with get_conn() as conn: conn.execute("INSERT INTO collections (name) VALUES ('Public')") conn.commit() cid = conn.execute("SELECT MAX(id) FROM collections").fetchone()[0] conn.execute( "INSERT INTO collection_pages (collection_id, title) VALUES (?, ?)", (cid, "Page publique"), ) conn.commit() r = anon_csrf(client).get(f"/workspace/public/{cid}") assert r.status_code == 200, r.text[:200] assert "Page publique" in r.text # Une collection absente ne doit surtout pas exiger une session (401). r2 = anon_csrf(client).get("/workspace/public/424242") assert r2.status_code != 401 # ══════════════════════ P1-6 — favorites ══════════════════════ def test_favorites_roundtrip_uses_current_schema(client): from app.db import get_conn with get_conn() as conn: conn.execute( "INSERT INTO pages (workspace, title, content, content_format) " "VALUES ('WS','Fav page','','blocks')" ) conn.commit() page_id = conn.execute("SELECT MAX(id) FROM pages").fetchone()[0] r = client.get("/workspace/favorites") assert r.status_code == 200, r.text assert r.json()["favorites"] == [] r = client.post("/workspace/favorites", json={"page_id": page_id}) assert r.status_code == 200, r.text r = client.get("/workspace/favorites") assert r.status_code == 200, r.text favs = r.json()["favorites"] assert len(favs) == 1 assert favs[0]["page_id"] == page_id assert favs[0]["page_title"] == "Fav page" assert client.delete(f"/workspace/favorites/{favs[0]['id']}").status_code == 200 assert client.get("/workspace/favorites").json()["favorites"] == [] def test_favorites_add_requires_page_id(client): assert client.post("/workspace/favorites", json={}).status_code == 400 # ══════════════════════ P1-7 — ordre de routes agents ══════════════════════ def test_agents_conversations_route_is_not_shadowed(client): r = client.get("/api/v2/agents/conversations") # 401 « API token required » = la route a bien été atteinte (avant : 422 # int_parsing sur /agents/{agent_id}). assert r.status_code == 401, r.text assert "int_parsing" not in r.text # ══════════════════════ P1-8 / P1-9 — synced blocks ══════════════════════ def test_page_editor_uses_board_prefix_for_synced_blocks(): from pathlib import Path src = Path("static/js/page_editor_scripts.js").read_text(encoding="utf-8") assert "'/api/synced-blocks'" not in src assert src.count("'/board/api/synced-blocks'") == 1 assert "'/board/api/synced-blocks/'" in src assert "Settings → Synced Blocks" not in src def test_synced_blocks_anonymous_is_401_not_500(client): c = anon_csrf(client) assert c.get("/board/api/synced-blocks").status_code == 401 assert c.get("/board/api/synced-blocks/1").status_code == 401 assert c.post("/board/api/synced-blocks", json={"title": "x"}).status_code == 401 def test_synced_block_cannot_be_edited_by_another_user(client): r = client.post("/board/api/synced-blocks", json={"title": "Bloc", "content": []}) assert r.status_code == 200, r.text sid = r.json()["synced_block_id"] intruder = login_test_client(client, user_id=2, login="intruder", is_admin=0) assert intruder.put(f"/board/api/synced-blocks/{sid}", json={"title": "pwn"}).status_code == 403 assert intruder.delete(f"/board/api/synced-blocks/{sid}").status_code == 403 from app.services.synced_blocks import get_synced_block assert get_synced_block(sid)["title"] == "Bloc" owner = login_test_client(client, user_id=1, login="tester", is_admin=1) assert owner.put(f"/board/api/synced-blocks/{sid}", json={"title": "ok"}).status_code == 200 assert get_synced_block(sid)["title"] == "ok" def test_synced_blocks_list_with_session(client): r = client.get("/board/api/synced-blocks") assert r.status_code == 200, r.text assert "synced_blocks" in r.json() # ══════════════════════ P1-5 — bouton Home ══════════════════════ def test_undefined_template_variable_is_logged_not_silent(caplog): """P1-10 : une variable absente du contexte rend ``""`` (contrattenu par 40+ templates optionnelles) mais ne doit plus être totally silencieuse.""" import logging from app.templating import ENV with caplog.at_level(logging.WARNING, logger="app.templating"): out = ENV.from_string("{{ totally_unknown_variable }}").render() assert out == "" assert any( "totally_unknown_variable" in r.getMessage() for r in caplog.records ), [r.getMessage() for r in caplog.records] def test_home_button_targets_the_active_workspace(client): """``local_workspaces[0]`` = premier workspace TRIÉ PAR NOM ; le Home doit cibler l'espace actif (``local_ws_id``).""" from pathlib import Path tpl = Path("app/templates/base.html").read_text(encoding="utf-8") m = re.search(r'