# tests/test_file_chat.py — Chat par fichier (#169)
"""Store + REST routes for the per-file chat history."""

import json
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": [], "read": {}}

    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": [], "read": {}}

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

    def test_preview_image_is_proxied_same_origin(self, monkeypatch):
        """BUG-109: a remote og:image is blocked by ``img-src 'self'``.

        A *relative* og:image must be resolved against the page and downloaded
        into ``chat_uploads`` so the browser can load it same-origin.
        """
        monkeypatch.setattr(_store, "_PREVIEW_CACHE", {})
        from backend.tools import web as _web

        monkeypatch.setattr(_web, "_assert_public_http_url", lambda url: None)

        class _Page:
            status_code = 200
            text = (
                '<meta property="og:title" content="Titre">'
                '<meta property="og:image" content="/img/pic.png">'
            )

        class _Img:
            status_code = 200
            content = b"\x89PNG-fake-bytes"
            headers = {"content-type": "image/png"}

        def _fake_get(url, **kw):
            return _Img() if url.endswith(".png") else _Page()

        monkeypatch.setattr(_store.httpx, "get", _fake_get)
        preview = _store.build_preview("voir https://site.test/a")
        assert preview["title"] == "Titre"
        assert preview["image"].startswith("/api/chat/attachment/")
        assert preview["image"].endswith(".png")
        stored = _store.UPLOAD_DIR / preview["image"].rsplit("/", 1)[1]
        assert stored.exists()
        assert stored.read_bytes() == b"\x89PNG-fake-bytes"

    def test_preview_image_failure_keeps_the_card(self, monkeypatch):
        """A dead thumbnail degrades to an empty image, never a lost card."""
        monkeypatch.setattr(_store, "_PREVIEW_CACHE", {})
        from backend.tools import web as _web

        monkeypatch.setattr(_web, "_assert_public_http_url", lambda url: None)

        class _Page:
            status_code = 200
            text = (
                '<meta property="og:title" content="Titre">'
                '<meta property="og:image" content="https://cdn.test/broken.jpg">'
            )

        class _Dead:
            status_code = 404
            content = b""
            headers = {}

        def _fake_get(url, **kw):
            return _Dead() if url.endswith(".jpg") else _Page()

        monkeypatch.setattr(_store.httpx, "get", _fake_get)
        preview = _store.build_preview("voir https://site.test/b")
        assert preview["title"] == "Titre"
        assert preview["image"] == ""


# --- #192 : accusé de réception ---------------------------------------------

class TestReadReceipts:
    def test_mark_read_roundtrip(self, tmp_path, monkeypatch):
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        assert _store.get_read("V", "a.md") == {}
        read = _store.mark_read("V", "a.md", "bruno")
        assert read["bruno"] > 0
        assert _store.get_read("V", "a.md") == read

    def test_mark_read_preserves_messages(self, tmp_path, monkeypatch):
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        _store.add_message("V", "a.md", "alice", "salut")
        _store.mark_read("V", "a.md", "bruno")
        assert [m["text"] for m in _store.get_messages("V", "a.md")] == ["salut"]
        assert "bruno" in _store.get_read("V", "a.md")

    def test_post_read_records_for_the_caller(self, client, test_vault_dir):
        r = client.post("/api/chat/read", json={"vault": "__global__", "path": "general"})
        assert r.status_code == 200, r.text
        read = r.json()["read"]
        assert read["anonymous"] > 0  # auth disabled in tests → fake user
        # the history endpoint exposes the same map (#192)
        assert client.get("/api/chat").json()["read"] == read

    def test_post_read_refuses_a_dm_we_are_not_in(self, client, test_vault_dir):
        r = client.post("/api/chat/read", json={"vault": "__dm__", "path": "alice|bob"})
        assert r.status_code == 403

    def test_post_read_requires_vault_and_path(self, client, test_vault_dir):
        assert client.post("/api/chat/read", json={}).status_code == 400

    def test_post_read_follows_vault_acl(self, client, test_vault_dir):
        r = client.post("/api/chat/read", json={"vault": "TestVault", "path": "note1.md"})
        assert r.status_code == 200, r.text
        assert "anonymous" in r.json()["read"]

    def test_post_read_broadcasts_chat_read(self, client, test_vault_dir, monkeypatch):
        from backend.routers import file_chat as _router

        events = []

        class _Spy:
            async def broadcast(self, event, data):
                events.append((event, data))

        monkeypatch.setattr(_router, "sse_manager", _Spy())
        client.post("/api/chat/read", json={"vault": "__global__", "path": "general"})
        assert [e for e, _ in events] == ["chat_read"]
        assert events[0][1]["user"] == "anonymous"
        assert events[0][1]["vault"] == "__global__"


# ═══════════════════════════════════════════════════════════════════
# #193 — un post est rendu comme un document markdown
# ═══════════════════════════════════════════════════════════════════

class TestMarkdownRendering:
    """Le post porte son HTML (mistune + sanitizer), comme une prévisualisation
    de document : le client n'a plus à réimplémenter le markdown."""

    def test_message_carries_rendered_html(self, tmp_path, monkeypatch):
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        msg = _store.add_message("V", "a.md", "u", "**gras** et `code`")
        assert "<strong>gras</strong>" in msg["html"]
        assert "<code>code</code>" in msg["html"]
        got = _store.get_messages("V", "a.md")
        assert "<strong>gras</strong>" in got[0]["html"]

    def test_html_is_not_persisted(self, tmp_path, monkeypatch):
        """`html` est recalculé à chaque lecture, jamais écrit dans le store."""
        chat_dir = tmp_path / "chats"
        monkeypatch.setattr(_store, "CHAT_DIR", chat_dir)
        _store.add_message("V", "a.md", "u", "**gras**")
        raw = json.loads(next(chat_dir.glob("*.json")).read_text(encoding="utf-8"))
        assert "html" not in raw["messages"][0]

    def test_fenced_code_block_keeps_its_language(self, tmp_path, monkeypatch):
        """La classe `language-xxx` est ce sur quoi hljs se branche (comme dans
        le viewer)."""
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        msg = _store.add_message("V", "a.md", "u", "```python\nprint(1)\n```")
        assert 'class="language-python"' in msg["html"]

    def test_tables_and_lists_are_rendered(self, tmp_path, monkeypatch):
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        msg = _store.add_message("V", "a.md", "u", "- un\n- deux")
        assert "<ul>" in msg["html"] and "<li>un</li>" in msg["html"]

    def test_raw_html_is_sanitized(self, tmp_path, monkeypatch):
        """Pas de XSS par le chat : le sanitizer des documents s'applique."""
        monkeypatch.setattr(_store, "CHAT_DIR", tmp_path / "chats")
        msg = _store.add_message("V", "a.md", "u", "<script>alert(1)</script>ok")
        assert "<script" not in msg["html"]

    def test_routes_return_html(self, client, test_vault_dir):
        r = client.post(
            "/api/file/TestVault/chat",
            json={"path": "note1.md", "text": "## Titre\n\n- item"},
        )
        assert r.status_code == 200
        html = r.json()["message"]["html"]
        assert "<h2" in html and "<li>item</li>" in html
        r2 = client.get("/api/file/TestVault/chat", params={"path": "note1.md"})
        assert "<h2" in r2.json()["messages"][0]["html"]

    def test_sse_broadcast_carries_html(self, client, test_vault_dir, monkeypatch):
        """L'écho SSE part avec le html : les autres clients n'attendent pas un
        refresh pour voir le rendu."""
        from backend.routers import file_chat as _router

        events = []

        class _Spy:
            async def broadcast(self, event, data):
                events.append((event, data))

        monkeypatch.setattr(_router, "sse_manager", _Spy())
        client.post("/api/chat", json={"text": "**fort**"})
        assert events and events[0][0] == "chat_message"
        assert "<strong>fort</strong>" in events[0][1]["message"]["html"]