gogo-telefon/tests/test_chat_widget.py

"""Chat widget e2e (M3): WebSocket booking flow → request in DB + proposal email.

Uses Starlette's sync TestClient for WebSocket support; the scripted LLM drives
the same agent loop as production.
"""

import pytest
from starlette.testclient import TestClient
from starlette.websockets import WebSocketDisconnect

from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse, set_llm
from gogo.main import create_app


def make_client() -> TestClient:
    return TestClient(create_app())


@pytest.fixture(autouse=True)
def fast_chat(monkeypatch):
    """Scripted LLM answers instantly — disable the per-second message gate."""
    from gogo.chat import router as chat_router

    monkeypatch.setattr(chat_router, "MIN_SECONDS_BETWEEN_MESSAGES", 0.0)


async def test_widget_js_served(gogo_client, session):
    resp = await gogo_client.get("/widget.js")
    assert resp.status_code == 200
    assert "gogo-bubble" in resp.text
    assert resp.headers["content-type"].startswith("application/javascript")


def test_ws_rejects_unknown_key(session, tenant):
    client = make_client()
    # server closes before accepting → the connect itself raises with our code
    with pytest.raises(WebSocketDisconnect) as exc:  # noqa: SIM117
        with client.websocket_connect("/widget/ws?key=nepostojeci"):
            pass
    assert exc.value.code == 4404


def test_ws_booking_flow_creates_request_and_email(session, tenant, emails, sms):
    """Full widget booking: greeting → messages → submit_booking_request →
    request stored, proposal email sent, chat transcript persisted."""
    import anyio
    from sqlalchemy import select

    from gogo.models import BookingRequest, ChatMessage

    set_llm(
        ScriptedLLM(
            [
                LLMResponse(text="Može! Koji je vaš broj telefona i ime?"),
                LLMResponse(
                    text="",
                    tool_uses=[
                        ToolUse(
                            id="t1",
                            name="submit_booking_request",
                            input={
                                "service_name_raw": "manikir",
                                "client_name": "Lejla",
                                "client_phone": "+38761555444",
                                "time_preference_text": "petak poslijepodne",
                                "summary": "Manikir, petak poslijepodne.",
                            },
                        )
                    ],
                    stop_reason="tool_use",
                ),
                LLMResponse(
                    text="Zahtjev sam proslijedila salonu — kontaktiraće vas radi potvrde!"
                ),
            ]
        )
    )
    try:
        client = make_client()
        with client.websocket_connect(f"/widget/ws?key={tenant.widget_public_key}") as ws:
            greeting = ws.receive_json()
            assert greeting["type"] == "greeting"
            assert "Gogo" in greeting["text"]
            resume_token = greeting["session"]
            assert resume_token

            ws.send_json({"type": "message", "text": "Može manikir u petak?", "website": ""})
            assert ws.receive_json()["type"] == "typing"
            reply = ws.receive_json()
            assert reply["type"] == "message"
            assert "broj telefona" in reply["text"]

            ws.send_json({"type": "message", "text": "Lejla, 061 555 444"})
            assert ws.receive_json()["type"] == "typing"
            reply2 = ws.receive_json()
            assert "proslijedila" in reply2["text"]

        # verify persisted state via a fresh event loop session
        async def check():
            import gogo.db as db

            async with db.get_sessionmaker()() as s:
                req = (await s.execute(select(BookingRequest))).scalar_one()
                assert req.client_name == "Lejla"
                assert req.source == "chat"
                assert req.chat_session_id is not None
                msgs = (await s.execute(select(ChatMessage))).scalars().all()
                roles = [m.role for m in msgs]
                assert roles.count("user") == 2
                assert roles.count("assistant") >= 3  # greeting + 2 replies

        anyio.run(check)
        assert len(emails) == 1  # proposal email to the owner
        assert "Lejla" in emails[0].subject
    finally:
        set_llm(None)


def test_ws_honeypot_drops_message(session, tenant, emails):
    set_llm(ScriptedLLM([LLMResponse(text="NE BI SMIO STIĆI")]))
    try:
        client = make_client()
        with client.websocket_connect(f"/widget/ws?key={tenant.widget_public_key}") as ws:
            ws.receive_json()  # greeting
            # bot fills the hidden field → silently ignored
            ws.send_json({"type": "message", "text": "spam", "website": "http://spam.example"})
            # then a real empty check: send another honeypot message; no replies expected
            ws.send_json({"type": "message", "text": "spam2", "website": "x"})
            # close cleanly — if the agent had run, ScriptedLLM would have been consumed
        assert True
    finally:
        set_llm(None)


def test_ws_resume_restores_history(session, tenant, emails):
    set_llm(ScriptedLLM([LLMResponse(text="Prvi odgovor.")]))
    try:
        client = make_client()
        with client.websocket_connect(f"/widget/ws?key={tenant.widget_public_key}") as ws:
            token = ws.receive_json()["session"]
            ws.send_json({"type": "message", "text": "Zdravo!"})
            ws.receive_json()  # typing
            ws.receive_json()  # reply

        with client.websocket_connect(
            f"/widget/ws?key={tenant.widget_public_key}&resume={token}"
        ) as ws:
            data = ws.receive_json()
            assert data["type"] == "resume"
            texts = [m["text"] for m in data["messages"]]
            assert "Zdravo!" in texts
            assert "Prvi odgovor." in texts
    finally:
        set_llm(None)