webmail-client/backend/tests/api/test_spellcheck.py

"""Spelling/grammar check proxy (compose window -> LanguageTool).

LanguageTool is replaced by a stub transport, so these run without the engine.
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from typing import Any

import httpx
import pytest
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.api import spellcheck
from app.config import get_settings
from app.identity.passwords import hash_password
from app.identity.sessions import create_session
from app.main import create_app
from app.models import User

LT_REPLY: dict[str, Any] = {
    "language": {
        "code": "auto",
        "name": "Auto",
        "detectedLanguage": {"code": "pl-PL", "name": "Polish"},
    },
    "matches": [
        {
            "offset": 12,
            "length": 6,
            "message": "Wyraz „prosze” to najczęściej literówka",
            "shortMessage": "Literówka",
            "replacements": [{"value": "proszę"}],
            "rule": {"id": "PL_PROSZE", "issueType": "misspelling"},
        },
        {
            "offset": 40,
            "length": 11,
            "message": "Powtórzony wyraz",
            "shortMessage": "",
            "replacements": [{"value": v} for v in "abcdefgh"],
            "rule": {"id": "WORD_REPEAT", "issueType": "duplication"},
        },
    ],
}


@pytest_asyncio.fixture
async def client(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[httpx.AsyncClient]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    session = async_sessionmaker(engine, expire_on_commit=False)()
    user = User(handle=f"sp{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
    session.add(user)
    await session.commit()
    token = await create_session(user.id, "pytest")
    async with httpx.AsyncClient(
        transport=ASGITransport(app=create_app()),
        base_url="http://testserver",
        cookies={"session": token, "csrf": "c"},
        headers={"X-CSRF": "c"},
    ) as c:
        yield c
    await session.execute(delete(User).where(User.id == user.id))
    await session.commit()
    await session.close()
    await engine.dispose()


def _stub(monkeypatch: pytest.MonkeyPatch, handler: Any) -> list[httpx.Request]:
    """Route the module's outbound calls to `handler`; return what was sent."""
    seen: list[httpx.Request] = []
    real = httpx.AsyncClient

    def wrapped(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        return handler(request)

    monkeypatch.setattr(
        spellcheck.httpx,
        "AsyncClient",
        lambda **kw: real(transport=httpx.MockTransport(wrapped), **kw),
    )
    monkeypatch.setattr(get_settings(), "languagetool_url", "http://lt.test:8010")
    return seen


async def test_check_returns_trimmed_typed_issues(
    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    sent = _stub(monkeypatch, lambda _r: httpx.Response(200, json=LT_REPLY))
    resp = await client.post(
        "/api/spellcheck",
        json={"text": "Dzień dobry prosze o pliki ważne ważne", "language": "auto"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert (body["language"], body["language_name"]) == ("pl-PL", "Polish")
    first, second = body["issues"]
    assert first["kind"] == "spelling" and first["replacements"] == ["proszę"]
    assert second["kind"] == "grammar" and len(second["replacements"]) == 6
    assert b"language=auto" in sent[0].content


async def test_unconfigured_or_down_engine_is_a_503(
    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    def down(_r: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("refused")

    _stub(monkeypatch, down)
    resp = await client.post("/api/spellcheck", json={"text": "hello"})
    assert resp.json()["error"]["code"] == "spellcheck_unavailable"

    monkeypatch.setattr(get_settings(), "languagetool_url", "")
    resp = await client.post("/api/spellcheck", json={"text": "hello"})
    assert resp.status_code == 503


async def test_empty_text_never_reaches_the_engine(
    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    sent = _stub(monkeypatch, lambda _r: httpx.Response(200, json=LT_REPLY))
    resp = await client.post("/api/spellcheck", json={"text": "   \n "})
    assert resp.json()["issues"] == [] and sent == []


async def test_settings_store_the_spellcheck_choice(client: httpx.AsyncClient) -> None:
    out = (await client.patch("/api/settings", json={"spellcheck_language": "pl-PL"})).json()
    assert out["spellcheck_language"] == "pl-PL" and out["spellcheck_enabled"] is True
    out = (await client.patch("/api/settings", json={"spellcheck_enabled": False})).json()
    assert out["spellcheck_enabled"] is False
    bad = (await client.patch("/api/settings", json={"spellcheck_language": "<script>"})).json()
    assert bad["spellcheck_language"] == "pl-PL"  # rejected, previous value kept


def test_auto_text_is_split_into_language_sized_chunks() -> None:
    text = "Witam,\nprosze o przesłanie dokumentów do jutra, to bardzo ważne.\n\nHi\n\nThanks!"
    parts = spellcheck.chunks(text)
    # The greeting rides with the long line after it; blank lines end a chunk.
    assert [t for _, t in parts] == [
        "Witam,\nprosze o przesłanie dokumentów do jutra, to bardzo ważne.",
        "Hi",
        "Thanks!",
    ]
    # Offsets point back into the original text.
    assert all(text[o : o + len(t)] == t for o, t in parts)


async def test_auto_checks_each_chunk_and_shifts_offsets(
    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    def per_chunk(request: httpx.Request) -> httpx.Response:
        body = request.content.decode()
        lang = ("pl-PL", "Polish") if "Witam" in body else ("en-US", "English (US)")
        return httpx.Response(
            200,
            json={
                "language": {"detectedLanguage": {"code": lang[0], "name": lang[1]}},
                "matches": [
                    {
                        "offset": 0,
                        "length": 2,
                        "message": "m",
                        "replacements": [],
                        "rule": {"id": "R", "issueType": "misspelling"},
                    }
                ],
            },
        )

    sent = _stub(monkeypatch, per_chunk)
    text = "Witam, to jest długi polski akapit pełen słów i zdań.\n\nHello there friend"
    out = (await client.post("/api/spellcheck", json={"text": text})).json()
    assert len(sent) == 2
    assert [i["offset"] for i in out["issues"]] == [0, text.index("Hello")]
    assert out["language_name"] == "Polish"  # most of the text