kamidso-back/tests/kamidso/test_conversation_service.py

import pytest

from kamidso.application.conversation_service import ConversationService, derive_working_title
from kamidso.contracts.errors import KamidsoError, KamidsoErrorCode
from kamidso.contracts.resource import ResourceRef
from kamidso.contracts.user import CurrentUser
from kamidso.repositories.store import RepositoryStore


@pytest.mark.asyncio
async def test_get_or_create_provisions_unknown_client_conversation_id():
    store = RepositoryStore()
    service = ConversationService(store)
    user = CurrentUser(
        id="usr_1",
        email="admin@kamidso.test",
        name="Admin",
        role="developer",
        workspace_id="ws_1",
    )
    await store.create_workspace({"_id": "ws_1", "organization_id": "org_1", "name": "Main"})

    client_id = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    conversation = await service.get_or_create(user, client_id)

    assert conversation.id == client_id
    stored = await service.get(user, client_id)
    assert stored is not None
    assert stored.id == client_id


@pytest.mark.asyncio
async def test_get_or_create_rejects_foreign_workspace_conversation():
    store = RepositoryStore()
    service = ConversationService(store)
    user_a = CurrentUser(
        id="usr_a",
        email="a@kamidso.test",
        name="A",
        role="developer",
        workspace_id="ws_a",
    )
    user_b = CurrentUser(
        id="usr_b",
        email="b@kamidso.test",
        name="B",
        role="developer",
        workspace_id="ws_b",
    )
    await store.create_workspace({"_id": "ws_a", "organization_id": "org_1", "name": "A"})
    await store.create_workspace({"_id": "ws_b", "organization_id": "org_1", "name": "B"})

    foreign = await service.create(user_a, conversation_id="conv_shared")

    with pytest.raises(KamidsoError) as exc:
        await service.get_or_create(user_b, foreign.id)

    assert exc.value.code == KamidsoErrorCode.WORKSPACE_ISOLATION


def test_derive_working_title_from_message_and_files():
    assert derive_working_title("تحلیل فروش ماهانه") == "تحلیل فروش ماهانه"
    assert derive_working_title(
        "فایل را تحلیل کن.",
        [ResourceRef(id="file_1", type="file", name="فروش ۱۴۰۳")],
    ) == "فروش ۱۴۰۳"


@pytest.mark.asyncio
async def test_first_user_message_replaces_placeholder_title():
    store = RepositoryStore()
    service = ConversationService(store)
    user = CurrentUser(
        id="usr_1",
        email="admin@kamidso.test",
        name="Admin",
        role="developer",
        workspace_id="ws_1",
    )
    await store.create_workspace({"_id": "ws_1", "organization_id": "org_1", "name": "Main"})
    conversation = await service.create(user)
    assert conversation.title == "گفت‌وگوی جدید"

    await service.add_message(user, conversation.id, role="user", content="تحلیل فروش ماهانه")
    stored = await service.get(user, conversation.id)
    assert stored is not None
    assert stored.title == "تحلیل فروش ماهانه"

    await service.add_message(user, conversation.id, role="user", content="سؤال دوم")
    stored = await service.get(user, conversation.id)
    assert stored is not None
    assert stored.title == "تحلیل فروش ماهانه"


@pytest.mark.asyncio
async def test_agent_kind_messages_are_hidden_from_product_history():
    store = RepositoryStore()
    service = ConversationService(store)
    user = CurrentUser(
        id="usr_1",
        email="admin@kamidso.test",
        name="Admin",
        role="developer",
        workspace_id="ws_1",
    )
    await store.create_workspace({"_id": "ws_1", "organization_id": "org_1", "name": "Main"})
    conversation = await service.create(user)
    await service.add_message(user, conversation.id, role="user", content="سلام")
    await store.create_message(
        {
            "workspace_id": user.workspace_id,
            "conversation_id": conversation.id,
            "role": "assistant",
            "content": "trace",
            "kind": "agent",
            "conv_id": conversation.id,
        }
    )
    history = await service.list_messages(user, conversation.id)
    assert [item.content for item in history] == ["سلام"]


@pytest.mark.asyncio
async def test_replace_resources_overwrites_conversation_pins():
    store = RepositoryStore()
    service = ConversationService(store)
    user = CurrentUser(
        id="usr_1",
        email="admin@kamidso.test",
        name="Admin",
        role="developer",
        workspace_id="ws_1",
    )
    await store.create_workspace({"_id": "ws_1", "organization_id": "org_1", "name": "Main"})
    conversation = await service.create(user)
    await service.replace_resources(
        user,
        conversation.id,
        [ResourceRef(id="file_old", type="file", name="old.xlsx")],
    )
    await service.replace_resources(
        user,
        conversation.id,
        [
            ResourceRef(id="file_book1", type="file", name="Book1 (1).xlsx"),
            ResourceRef(id="ds_1", type="datasource", name="sales"),
        ],
    )
    stored = await service.get(user, conversation.id)
    assert stored is not None
    assert [(ref.id, ref.type, ref.name) for ref in stored.resource_refs] == [
        ("file_book1", "file", "Book1 (1).xlsx"),
        ("ds_1", "datasource", "sales"),
    ]