367 lines
16 KiB
Python
367 lines
16 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
|
|
|
|
|
|
def _login(client, username="admin", password="chab30") -> dict:
|
|
"""Log in against the auth-enabled ``admin_client`` fixture (#191)."""
|
|
resp = client.post("/api/auth/login", json={"username": username, "password": password})
|
|
assert resp.status_code == 200, resp.text
|
|
return {"Authorization": f"Bearer {resp.json()['access_token']}"}
|
|
|
|
|
|
@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"
|
|
|
|
|
|
# --- #191 : suppression, messages privés, link preview ----------------------
|
|
|
|
class TestDelete:
|
|
def test_author_deletes_own_post(self, client, test_vault_dir):
|
|
msg = client.post("/api/chat", json={"text": "a supprimer"}).json()["message"]
|
|
r = client.delete(f"/api/chat/{msg['id']}")
|
|
assert r.status_code == 200
|
|
assert client.get("/api/chat").json()["messages"] == []
|
|
|
|
def test_delete_unknown_404(self, client, test_vault_dir):
|
|
assert client.delete("/api/chat/nope").status_code == 404
|
|
|
|
def test_removed_message_cannot_be_deleted_twice(self, client, test_vault_dir):
|
|
msg = client.post("/api/chat", json={"text": "x"}).json()["message"]
|
|
client.delete(f"/api/chat/{msg['id']}")
|
|
assert client.delete(f"/api/chat/{msg['id']}").status_code == 404
|
|
|
|
|
|
class TestPrivateChat:
|
|
"""DM flows need a real logged-in identity (auth enabled)."""
|
|
|
|
def test_dm_roundtrip(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
r = admin_client.post("/api/chat/dm/normaluser", json={"text": "salut"}, headers=h)
|
|
assert r.status_code == 200
|
|
assert r.json()["message"]["text"] == "salut"
|
|
|
|
def test_dm_unknown_user_404(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
r = admin_client.post("/api/chat/dm/ghost", json={"text": "hi"}, headers=h)
|
|
assert r.status_code == 404
|
|
|
|
def test_dm_to_self_400(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
r = admin_client.post("/api/chat/dm/admin", json={"text": "hi"}, headers=h)
|
|
assert r.status_code == 400
|
|
|
|
def test_dm_isolated_from_general(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
admin_client.post("/api/chat", json={"text": "public"}, headers=h)
|
|
admin_client.post("/api/chat/dm/normaluser", json={"text": "prive"}, headers=h)
|
|
gen = admin_client.get("/api/chat", headers=h).json()["messages"]
|
|
assert [m["text"] for m in gen] == ["public"]
|
|
dm = admin_client.get("/api/chat/dm/normaluser", headers=h).json()["messages"]
|
|
assert [m["text"] for m in dm] == ["prive"]
|
|
|
|
def test_dm_history(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
admin_client.post("/api/chat/dm/normaluser", json={"text": "hystorique"}, headers=h)
|
|
r = admin_client.get("/api/chat/dm/normaluser", headers=h)
|
|
assert [m["text"] for m in r.json()["messages"]] == ["hystorique"]
|
|
|
|
def test_dm_pair_is_shared_not_duplicated(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
admin_client.post("/api/chat/dm/normaluser", json={"text": "a"}, headers=h)
|
|
assert _store.dm_path("admin", "normaluser") == _store.dm_path("normaluser", "admin")
|
|
assert len(admin_client.get("/api/chat/dm/normaluser", headers=h).json()["messages"]) == 1
|
|
|
|
def test_dm_delete(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
msg = admin_client.post(
|
|
"/api/chat/dm/normaluser", json={"text": "adieu"}, headers=h
|
|
).json()["message"]
|
|
r = admin_client.delete(f"/api/chat/dm/normaluser/{msg['id']}", headers=h)
|
|
assert r.status_code == 200
|
|
assert admin_client.get("/api/chat/dm/normaluser", headers=h).json()["messages"] == []
|
|
|
|
def test_requires_auth(self, admin_client, test_vault_dir):
|
|
assert admin_client.get("/api/chat/dm/normaluser").status_code == 401
|
|
assert admin_client.get("/api/chat/users").status_code == 401
|
|
|
|
def test_user_list_for_picker(self, admin_client, test_vault_dir):
|
|
h = _login(admin_client)
|
|
r = admin_client.get("/api/chat/users", headers=h)
|
|
assert r.status_code == 200
|
|
names = [u["username"] for u in r.json()]
|
|
assert "normaluser" in names
|
|
assert "admin" not in names # oneself is not a DM target
|
|
assert all("password" not in u and "hash" not in str(u) for u in r.json())
|
|
|
|
|
|
class TestLinkPreview:
|
|
def test_no_url_no_preview(self, client, test_vault_dir, monkeypatch):
|
|
monkeypatch.setattr(_store, "build_preview", lambda text: None)
|
|
msg = client.post("/api/chat", json={"text": "pas d url"}).json()["message"]
|
|
# the schema always serialises the key (null) — it must stay falsy
|
|
assert not msg.get("preview")
|
|
|
|
def test_preview_attached_when_fetched(self, client, test_vault_dir, monkeypatch):
|
|
monkeypatch.setattr(
|
|
_store, "build_preview",
|
|
lambda text: {"url": "https://x.test", "title": "Titre", "description": "", "image": "", "site": "x.test"},
|
|
)
|
|
msg = client.post("/api/chat", json={"text": "voir https://x.test"}).json()["message"]
|
|
assert msg["preview"]["title"] == "Titre"
|
|
|
|
def test_preview_persisted_in_history(self, client, test_vault_dir, monkeypatch):
|
|
monkeypatch.setattr(
|
|
_store, "build_preview",
|
|
lambda text: {"url": "https://y.test", "title": "Y", "description": "", "image": "", "site": "y.test"},
|
|
)
|
|
client.post("/api/chat", json={"text": "https://y.test"})
|
|
got = client.get("/api/chat").json()["messages"]
|
|
assert got[-1]["preview"]["site"] == "y.test"
|
|
|
|
def test_build_preview_rejects_non_http(self):
|
|
assert _store.build_preview("javascript:alert(1)") is None
|
|
assert _store.build_preview("aucune URL") is None
|
|
|
|
def test_build_preview_blocks_private_hosts(self, monkeypatch):
|
|
# SSRF guard rejects loopback before any network round-trip
|
|
monkeypatch.setattr(_store, "_PREVIEW_CACHE", {})
|
|
assert _store.build_preview("http://127.0.0.1/x") is None
|
|
|
|
def test_build_preview_happy_path_parses_og(self, monkeypatch):
|
|
"""The fetch itself (a bad kwarg here used to fail silently)."""
|
|
monkeypatch.setattr(_store, "_PREVIEW_CACHE", {})
|
|
# the SSRF guard resolves DNS for real — out of scope for this test
|
|
from backend.tools import web as _web
|
|
|
|
monkeypatch.setattr(_web, "_assert_public_http_url", lambda url: None)
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = (
|
|
"<html><head><title>Fallback</title>"
|
|
'<meta property="og:title" content="Titre OG">'
|
|
'<meta property="og:description" content="Desc">'
|
|
'<meta property="og:site_name" content="Site">'
|
|
"</head></html>"
|
|
)
|
|
|
|
called = {}
|
|
|
|
def _fake_get(url, **kw):
|
|
called["url"] = url
|
|
return _Resp()
|
|
|
|
monkeypatch.setattr(_store.httpx, "get", _fake_get)
|
|
preview = _store.build_preview("voir https://example.test/page")
|
|
assert called["url"] == "https://example.test/page"
|
|
assert preview == {
|
|
"url": "https://example.test/page",
|
|
"title": "Titre OG",
|
|
"description": "Desc",
|
|
"image": "",
|
|
"site": "Site",
|
|
}
|