CODE_ASSISTANT/5_django_backend/tests/test_sessions.py

"""세션 미러 API — snap.api.ts 계약. OpenCode 는 가짜로."""

import httpx
import pytest

from apps.chat.models import ChatMessage, ChatSession

SESSION_KEYS = {"id", "title", "titleLlm", "isGenerating", "createdAt", "updatedAt"}
MESSAGE_KEYS = {"sessionId", "role", "content", "createdAt", "inputTokens", "outputTokens", "costUsd", "elapsedMs"}


class FakeOpencode:
    def __init__(self):
        self.n = 0
        self.aborted = []
        self.down = False

    def create_session(self):
        if self.down:
            raise httpx.ConnectError("refused")
        self.n += 1
        return {"id": f"ses_{self.n}", "title": ""}

    def abort_session(self, sid):
        self.aborted.append(sid)
        return True


@pytest.fixture
def oc(monkeypatch):
    fake = FakeOpencode()
    monkeypatch.setattr("apps.chat.views.opencode_service", fake)
    return fake


def _mk(user, sid="ses_a", **kw):
    return ChatSession.objects.create(id=sid, user=user, **kw)


def test_create_session_mirrors_opencode(auth_api, oc):
    res = auth_api.post("/api/v1/chat/sessions", {}, format="json")
    assert res.status_code == 201
    data = res.json()["data"]
    assert set(data) == SESSION_KEYS
    assert data["id"] == "ses_1" and data["isGenerating"] is False and data["titleLlm"] is None
    assert ChatSession.objects.get(id="ses_1").user.email == "u@x.com"


def test_create_session_when_opencode_down_503(auth_api, oc):
    oc.down = True
    res = auth_api.post("/api/v1/chat/sessions", {}, format="json")
    assert res.status_code == 503
    assert res.json()["code"] == "UPSTREAM_UNAVAILABLE"


def test_list_paginated_newest_first_own_only(auth_api, user, django_user_model):
    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
    for i in range(5):
        _mk(user, f"ses_{i}")
    _mk(other, "ses_other")
    res = auth_api.get("/api/v1/chat/sessions?page=1&limit=3")
    body = res.json()
    assert [s["id"] for s in body["data"]] == ["ses_4", "ses_3", "ses_2"]
    assert body["counts"] == 5
    assert body["meta"] == {
        "currentPage": 1,
        "pageSize": 3,
        "totalItems": 5,
        "totalPages": 2,
        "hasNextPage": True,
        "hasPreviousPage": False,
    }
    res2 = auth_api.get("/api/v1/chat/sessions?page=2&limit=3")
    assert [s["id"] for s in res2.json()["data"]] == ["ses_1", "ses_0"]


def test_messages_detail(auth_api, user):
    s = _mk(user, title_llm="첫 질문")
    ChatMessage.objects.create(session=s, role="user", content="hi")
    ChatMessage.objects.create(
        session=s, role="assistant", content="hello", input_tokens=10, output_tokens=5, cost_usd=0.001, elapsed_ms=1200
    )
    res = auth_api.get("/api/v1/chat/sessions/ses_a/messages")
    data = res.json()["data"]
    assert set(data) == SESSION_KEYS | {"messages"}
    assert data["titleLlm"] == "첫 질문"
    assert [m["role"] for m in data["messages"]] == ["user", "assistant"]
    assert set(data["messages"][1]) == MESSAGE_KEYS
    assert data["messages"][1]["inputTokens"] == 10
    assert data["messages"][0]["inputTokens"] is None


def test_messages_of_other_user_404(auth_api, django_user_model):
    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
    _mk(other, "ses_other")
    assert auth_api.get("/api/v1/chat/sessions/ses_other/messages").status_code == 404


def test_search_messages_icontains_own_only(auth_api, user, django_user_model):
    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
    mine = _mk(user, "ses_m")
    theirs = _mk(other, "ses_t")
    ChatMessage.objects.create(session=mine, role="user", content="SELECT * FROM mara")
    ChatMessage.objects.create(session=mine, role="assistant", content="MARA 는 자재 마스터")
    ChatMessage.objects.create(session=mine, role="user", content="다른 얘기")
    ChatMessage.objects.create(session=theirs, role="user", content="mara 남의 것")
    res = auth_api.get("/api/v1/chat/sessions/search?query=mara&page=1&limit=20")
    body = res.json()
    assert body["counts"] == 2
    assert {m["content"] for m in body["data"]} == {"SELECT * FROM mara", "MARA 는 자재 마스터"}
    assert all(m["sessionId"] == "ses_m" for m in body["data"])


def test_search_empty_query_400(auth_api):
    assert auth_api.get("/api/v1/chat/sessions/search?query=%20").status_code == 400


def test_cancel_aborts_and_clears_generating(auth_api, user, oc):
    _mk(user, is_generating=True)
    res = auth_api.post("/api/v1/chat/sessions/ses_a/cancel", {}, format="json")
    assert res.status_code == 200 and res.json()["data"] is None
    assert oc.aborted == ["ses_a"]
    assert ChatSession.objects.get(id="ses_a").is_generating is False