Files
ObsiGate/tests/test_file_chat.py
T
bruno ae06436f91
CI / lint (push) Successful in 2m48s
CI / security (push) Successful in 1m34s
CI / test (push) Successful in 4m33s
CI / build (push) Successful in 1m30s
CI / e2e (push) Successful in 17m5s
feat: chat — suivi du document, chat général en onglet sidebar, onglets au-dessus du filtre #190
- **A — Suivi du document** : l'en-tête du panneau chat affiche le document
  ciblé (titre + chemin) et `followFileChat()` appelé par `renderFile()`
  re-cible le panneau ouvert à chaque changement de document.
- **B — Chat général** : onglet « Chat » (dernier) dans la sidebar,
  conversation centrale stockée dans le store #169 via les sentinelles
  `__global__/general` (`GET/POST /api/chat`) ; pastille de messages non
  lus sur l'onglet ; pièces jointes image/vidéo (`POST /api/chat/upload` :
  allow-list d'extensions, 25 MB, nom UUID ; `GET /api/chat/attachment/{name}`
  résolu contre l'allow-list) ; URL cliquables au rendu ; date/heure d'envoi.
- **C — Ordre des onglets** : la barre de filtre passe sous les onglets et
  sert de recherche dans le chat (texte + auteur) quand l'onglet Chat est
  actif (`switchSidebarTab` + `routeFilter`).
- Transport : broadcast SSE `chat_message` réutilisé (`vault __global__`
  route vers la sidebar, sinon le panneau fichier) — pas de 2ᵉ WebSocket.
- i18n FR/EN (6 clés), CSS bloc #190, `initSidebarChat()` dans l'orchestrateur.
- Tests : pytest 27 chat (12 nouveaux) ; JSDOM `filechat.test.mjs` 11 (5
  nouveaux) ; suite 1655 passed, ruff/mypy 0, validate-imports 42 modules.
- Docs : CHANGELOG [Unreleased], ROADMAP #190 → index, fiche
  `file-chat-169.md` §#190, guide « Chat général », journal des interventions.
2026-10-08 20:40:54 -04:00

210 lines
9.3 KiB
Python

# tests/test_file_chat.py — Chat par fichier (#169)
"""Store + REST routes for the per-file chat history."""
import time
import pytest
from backend import file_chat as _store
@pytest.fixture(autouse=True)
def _isolated_chat_dir(tmp_path, monkeypatch):
"""Keep every test (store AND routes) out of the real ``data/chats``/
``data/chat_uploads``."""
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
monkeypatch.setattr(_store, "UPLOAD_DIR", tmp_path / "uploads")
class TestStore:
def test_get_messages_empty_when_missing(self, tmp_path, monkeypatch):
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
assert _store.get_messages("V", "a/b.md") == []
def test_add_then_get_roundtrip(self, tmp_path, monkeypatch):
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
msg = _store.add_message("V", "a/b.md", "bruno", "Bonjour !")
assert msg["user"] == "bruno"
assert msg["text"] == "Bonjour !"
assert msg["id"]
assert msg["ts"] <= time.time() + 1
got = _store.get_messages("V", "a/b.md")
assert len(got) == 1
assert got[0]["id"] == msg["id"]
def test_conversation_is_per_file(self, tmp_path, monkeypatch):
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
_store.add_message("V", "a.md", "u", "pour a")
_store.add_message("V", "b.md", "u", "pour b")
assert [m["text"] for m in _store.get_messages("V", "a.md")] == ["pour a"]
assert [m["text"] for m in _store.get_messages("V", "b.md")] == ["pour b"]
def test_path_traversal_never_escapes_chat_dir(self, tmp_path, monkeypatch):
"""The chat file name is hashed: '../../etc' cannot escape CHAT_DIR."""
chat_dir = tmp_path / "chats"
monkeypatch.setattr(_store, "CHAT_DIR", chat_dir)
_store.add_message("V", "../../etc/passwd", "u", "x")
files = list(chat_dir.iterdir())
assert len(files) == 1
assert files[0].parent == chat_dir
assert "/" not in files[0].name and "\\" not in files[0].name
def test_retention_cap_drops_oldest(self, tmp_path, monkeypatch):
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
n = _store.MAX_MESSAGES + 25
for i in range(n):
_store.add_message("V", "a.md", "u", f"m{i}")
got = _store.get_messages("V", "a.md")
assert len(got) == _store.MAX_MESSAGES
assert got[0]["text"] == f"m{25}" # first 25 dropped
assert got[-1]["text"] == f"m{n - 1}"
def test_text_truncated_to_max(self, tmp_path, monkeypatch):
monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
_store.add_message("V", "a.md", "u", "x" * (_store.MAX_TEXT + 500))
got = _store.get_messages("V", "a.md")
assert len(got[0]["text"]) == _store.MAX_TEXT
def test_corrupt_file_falls_back_to_empty(self, tmp_path, monkeypatch):
chat_dir = tmp_path / "chats"
monkeypatch.setattr(_store, "CHAT_DIR", chat_dir)
chat_dir.mkdir(parents=True)
target = _store._chat_file("V", "a.md")
target.write_text("{not json", encoding="utf-8")
assert _store.get_messages("V", "a.md") == []
class TestRoutes:
def test_history_requires_existing_vault(self, client, test_vault_dir):
r = client.get("/api/file/TestVault/chat", params={"path": "note1.md"})
assert r.status_code == 200
assert r.json() == {"messages": []}
def test_post_and_get_roundtrip(self, client, test_vault_dir):
r = client.post(
"/api/file/TestVault/chat",
json={"path": "note1.md", "text": "Vu hier ?"},
)
assert r.status_code == 200
body = r.json()
assert body["status"] == "ok"
assert body["message"]["text"] == "Vu hier ?"
r2 = client.get("/api/file/TestVault/chat", params={"path": "note1.md"})
assert [m["text"] for m in r2.json()["messages"]] == ["Vu hier ?"]
def test_post_requires_text(self, client, test_vault_dir):
r = client.post("/api/file/TestVault/chat", json={"path": "note1.md", "text": " "})
assert r.status_code == 400
def test_post_requires_path(self, client, test_vault_dir):
r = client.post("/api/file/TestVault/chat", json={"path": "", "text": "hi"})
assert r.status_code == 400
def test_unknown_vault_404(self, client, test_vault_dir):
r = client.get("/api/file/Nope/chat", params={"path": "x.md"})
assert r.status_code == 404
def test_path_traversal_rejected(self, client, test_vault_dir):
r = client.get(
"/api/file/TestVault/chat", params={"path": "../../etc/passwd"}
)
assert r.status_code in (403, 500)
r2 = client.post(
"/api/file/TestVault/chat",
json={"path": "../../etc/passwd", "text": "x"},
)
assert r2.status_code in (403, 500)
@pytest.mark.parametrize("path", ["note1.md", "sub/deep/file.md"])
def test_history_per_path_isolated(client, test_vault_dir, path):
client.post("/api/file/TestVault/chat", json={"path": path, "text": "m"})
other = "note1.md" if path != "note1.md" else "note2.md"
r = client.get("/api/file/TestVault/chat", params={"path": other})
assert r.json()["messages"] == []
# --- #190 : chat général + pièces jointes -----------------------------------
class TestGlobalChat:
def test_history_empty_by_default(self, client, test_vault_dir):
r = client.get("/api/chat")
assert r.status_code == 200
assert r.json() == {"messages": []}
def test_post_and_get_roundtrip(self, client, test_vault_dir):
r = client.post("/api/chat", json={"text": "Bonjour à tous"})
assert r.status_code == 200
msg = r.json()["message"]
assert msg["text"] == "Bonjour à tous"
assert msg["user"] # authenticated user recorded
got = client.get("/api/chat").json()["messages"]
assert [m["id"] for m in got] == [msg["id"]]
def test_empty_text_rejected(self, client, test_vault_dir):
assert client.post("/api/chat", json={"text": " "}).status_code == 400
def test_general_chat_isolated_from_file_chats(self, client, test_vault_dir):
client.post("/api/chat", json={"text": "général"})
client.post("/api/file/TestVault/chat", json={"path": "note1.md", "text": "fichier"})
assert len(client.get("/api/chat").json()["messages"]) == 1
assert len(client.get("/api/file/TestVault/chat", params={"path": "note1.md"}).json()["messages"]) == 1
def test_unknown_attachment_rejected(self, client, test_vault_dir):
r = client.post("/api/chat", json={
"text": "pièce jointe",
"attachment": {"name": "nope.png", "url": "javascript:alert(1)", "kind": "image"},
})
assert r.status_code == 400
def test_external_url_attachment_accepted(self, client, test_vault_dir):
r = client.post("/api/chat", json={
"text": "un lien",
"attachment": {"name": "", "url": "https://example.com/x.png", "kind": "image"},
})
assert r.status_code == 200
assert r.json()["message"]["attachment"]["url"] == "https://example.com/x.png"
class TestAttachments:
def test_upload_and_serve_roundtrip(self, client, test_vault_dir):
files = {"file": ("shot.png", b"\x89PNG\r\n\x1a\nfake", "image/png")}
r = client.post("/api/chat/upload", files=files)
assert r.status_code == 200
att = r.json()["attachment"]
assert att["kind"] == "image"
assert att["mime"] == "image/png"
# served back with the right content type
r2 = client.get(f"/api/chat/attachment/{att['name']}")
assert r2.status_code == 200
assert r2.headers["content-type"].startswith("image/png")
def test_video_kind_detection(self, client, test_vault_dir):
files = {"file": ("clip.mp4", b"\x00\x00\x00\x18ftyp", "video/mp4")}
att = client.post("/api/chat/upload", files=files).json()["attachment"]
assert att["kind"] == "video"
def test_extension_not_allowed(self, client, test_vault_dir):
files = {"file": ("evil.exe", b"MZ...", "application/octet-stream")}
assert client.post("/api/chat/upload", files=files).status_code == 400
def test_empty_file_rejected(self, client, test_vault_dir):
files = {"file": ("empty.png", b"", "image/png")}
assert client.post("/api/chat/upload", files=files).status_code == 400
def test_attachment_traversal_rejected(self, client, test_vault_dir):
# A traversal/unknown name must never produce a media response: the
# route 404s, or the SPA catch-all answers HTML (200 + text/html).
for bad in ("../secret.png", "sub/evil.png", "..%2Fsecret.png", "nope.png"):
r = client.get(f"/api/chat/attachment/{bad}")
ct = r.headers.get("content-type", "")
assert not ct.startswith(("image/", "video/")), (bad, r.status_code, ct)
assert r.status_code in (404, 400, 422, 200), bad
def test_message_with_uploaded_attachment(self, client, test_vault_dir):
files = {"file": ("pic.webp", b"RIFF0000WEBP", "image/webp")}
att = client.post("/api/chat/upload", files=files).json()["attachment"]
r = client.post("/api/chat", json={"text": "voici", "attachment": att})
assert r.status_code == 200
assert r.json()["message"]["attachment"]["kind"] == "image"