chat/apps/chat/tests/test_messages.py

import uuid
from unittest.mock import Mock

import pytest

from apps.chat.models import Conversation
from apps.chat.services.message import MessageService


def _make_service(*, mm_post_id="post-1", download_url=None, publishers=None):
    """Return a MessageService with all external deps mocked."""
    mm_client = Mock()
    mm_client.post_message.return_value = mm_post_id

    storage = Mock()
    storage.get_download_url.return_value = download_url

    return (
        MessageService(
            mattermost_client=mm_client,
            storage_service=storage,
            publishers=publishers if publishers is not None else [],
        ),
        mm_client,
        storage,
    )


@pytest.mark.django_db
def test_send_text_message():
    conv = Conversation.objects.create(mattermost_channel_id="ch-send-text")
    sender_uuid = uuid.uuid4()

    ws_publisher = Mock()
    svc, mm_client, storage = _make_service(mm_post_id="post-abc", publishers=[ws_publisher])

    result = svc.send(conv.uuid, sender_uuid, "text", text="Hello world")

    mm_client.post_message.assert_called_once_with("ch-send-text", "Hello world", sender_uuid)
    storage.get_download_url.assert_not_called()
    ws_publisher.publish.assert_called_once()

    event = ws_publisher.publish.call_args[0][0]
    assert event.post_id == "post-abc"
    assert event.sender_uuid == sender_uuid
    assert event.message_type == "text"

    assert result["post_id"] == "post-abc"
    assert result["text"] == "Hello world"
    assert result["url"] is None


@pytest.mark.django_db
def test_send_image_message():
    conv = Conversation.objects.create(mattermost_channel_id="ch-send-image")
    sender_uuid = uuid.uuid4()

    object_key = "images/abc123/photo.jpg"
    download_url = "http://minio.local/chat/images/abc123/photo.jpg?X-Amz-Signature=..."
    svc, mm_client, storage = _make_service(download_url=download_url, publishers=[])

    result = svc.send(conv.uuid, sender_uuid, "image", object_key=object_key)

    storage.get_download_url.assert_called_once_with(object_key)
    mm_client.post_message.assert_called_once_with(
        "ch-send-image", f"image:{object_key}", sender_uuid
    )
    assert result["url"] == download_url
    assert result["message_type"] == "image"


@pytest.mark.django_db
def test_list_messages_normalized():
    conv = Conversation.objects.create(mattermost_channel_id="ch-list")
    sender_uuid = uuid.uuid4()
    mm_user_id = "mm-service-bot"  # every post is authored by the service account

    object_key = "voice/def456/clip.ogg"
    raw_posts = [
        {
            "id": "p1",
            "user_id": mm_user_id,
            "message": "First message",
            "props": {"sender_uuid": str(sender_uuid)},
            "create_at": 1700000000000,
        },
        {
            "id": "p2",
            "user_id": mm_user_id,
            "message": f"voice:{object_key}",
            "props": {"sender_uuid": str(sender_uuid)},
            "create_at": 1700000001000,
        },
        {
            "id": "sys1",
            "type": "system_add_to_channel",
            "user_id": mm_user_id,
            "message": "user-x added to the channel",
            "create_at": 1700000002000,
        },
    ]

    download_url = "http://minio.local/chat/voice/def456/clip.ogg?X-Amz-Signature=..."
    mm_client = Mock()
    mm_client.get_posts.return_value = raw_posts
    svc, _, storage = _make_service(download_url=download_url, publishers=[])
    svc._mm = mm_client  # inject after construction to keep _make_service simple

    messages = svc.list_messages(conv.uuid, page=0, per_page=20)

    assert len(messages) == 3

    text_msg = messages[0]
    assert text_msg["post_id"] == "p1"
    assert text_msg["sender_uuid"] == sender_uuid
    assert text_msg["message_type"] == "text"
    assert text_msg["mattermost_message_type"] == ""
    assert text_msg["text"] == "First message"
    assert text_msg["url"] is None

    voice_msg = messages[1]
    assert voice_msg["post_id"] == "p2"
    assert voice_msg["sender_uuid"] == sender_uuid
    assert voice_msg["message_type"] == "voice"
    assert voice_msg["mattermost_message_type"] == ""
    assert voice_msg["url"] == download_url
    assert voice_msg["text"] is None

    storage.get_download_url.assert_called_once_with(object_key)

    system_msg = messages[2]
    assert system_msg["post_id"] == "sys1"
    assert system_msg["sender_uuid"] is None
    assert system_msg["mattermost_message_type"] == "system_add_to_channel"


@pytest.mark.django_db
def test_list_messages_treats_colon_in_plain_text_as_text():
    """A plain-text message that happens to contain a colon but doesn't use
    a recognized media-type prefix must not be mistaken for a media message.
    """
    conv = Conversation.objects.create(mattermost_channel_id="ch-list-colon")

    raw_posts = [
        {
            "id": "p1",
            "user_id": None,
            "message": "Meeting at 10:30",
            "create_at": 1700000000000,
        },
    ]

    mm_client = Mock()
    mm_client.get_posts.return_value = raw_posts
    svc, _, storage = _make_service(publishers=[])
    svc._mm = mm_client

    messages = svc.list_messages(conv.uuid, page=0, per_page=20)

    assert messages[0]["message_type"] == "text"
    assert messages[0]["text"] == "Meeting at 10:30"
    storage.get_download_url.assert_not_called()