mattermost/python-sdk/tests/test_posts_files_kv.py

# Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
# See LICENSE.txt for license information.

"""
Tests for Post, File, and KV Store API client methods.

This module tests the wrapper types and client method implementations for
Post, File, and KV Store operations.
"""

import pytest
from unittest.mock import MagicMock

import grpc


class TestPostWrapperTypes:
    """Tests for Post-related wrapper dataclasses."""

    def test_post_from_proto_and_to_proto(self):
        """Test Post wrapper round-trip conversion."""
        from mattermost_plugin._internal.wrappers import Post
        from mattermost_plugin.grpc import post_pb2

        # Create a protobuf Post
        proto_post = post_pb2.Post(
            id="post123",
            channel_id="channel123",
            user_id="user123",
            message="Hello, world!",
            create_at=1234567890000,
            update_at=1234567890000,
            is_pinned=True,
            root_id="",
            type="",
            hashtags="#test",
        )
        proto_post.file_ids.append("file1")
        proto_post.file_ids.append("file2")

        # Convert to wrapper
        post = Post.from_proto(proto_post)

        # Verify fields
        assert post.id == "post123"
        assert post.channel_id == "channel123"
        assert post.user_id == "user123"
        assert post.message == "Hello, world!"
        assert post.create_at == 1234567890000
        assert post.is_pinned is True
        assert post.hashtags == "#test"
        assert post.file_ids == ["file1", "file2"]

        # Convert back to proto
        proto_back = post.to_proto()

        # Verify round-trip
        assert proto_back.id == proto_post.id
        assert proto_back.channel_id == proto_post.channel_id
        assert proto_back.message == proto_post.message
        assert list(proto_back.file_ids) == ["file1", "file2"]

    def test_reaction_from_proto_and_to_proto(self):
        """Test Reaction wrapper round-trip conversion."""
        from mattermost_plugin._internal.wrappers import Reaction
        from mattermost_plugin.grpc import post_pb2

        # Create a protobuf Reaction
        proto_reaction = post_pb2.Reaction(
            user_id="user123",
            post_id="post123",
            emoji_name="thumbsup",
            create_at=1234567890000,
        )

        # Convert to wrapper
        reaction = Reaction.from_proto(proto_reaction)

        # Verify fields
        assert reaction.user_id == "user123"
        assert reaction.post_id == "post123"
        assert reaction.emoji_name == "thumbsup"
        assert reaction.create_at == 1234567890000

        # Convert back to proto
        proto_back = reaction.to_proto()

        # Verify round-trip
        assert proto_back.user_id == proto_reaction.user_id
        assert proto_back.post_id == proto_reaction.post_id
        assert proto_back.emoji_name == proto_reaction.emoji_name

    def test_post_list_from_proto(self):
        """Test PostList wrapper conversion."""
        from mattermost_plugin._internal.wrappers import PostList
        from mattermost_plugin.grpc import post_pb2

        # Create a protobuf PostList
        proto_post_list = post_pb2.PostList(
            next_post_id="next123",
            prev_post_id="prev123",
            has_next=True,
        )
        proto_post_list.order.append("post1")
        proto_post_list.order.append("post2")

        post1 = post_pb2.Post(id="post1", channel_id="ch1", message="First")
        post2 = post_pb2.Post(id="post2", channel_id="ch1", message="Second")
        proto_post_list.posts["post1"].CopyFrom(post1)
        proto_post_list.posts["post2"].CopyFrom(post2)

        # Convert to wrapper
        post_list = PostList.from_proto(proto_post_list)

        # Verify fields
        assert post_list.order == ["post1", "post2"]
        assert len(post_list.posts) == 2
        assert post_list.posts["post1"].message == "First"
        assert post_list.posts["post2"].message == "Second"
        assert post_list.next_post_id == "next123"
        assert post_list.has_next is True


class TestFileWrapperTypes:
    """Tests for File-related wrapper dataclasses."""

    def test_file_info_from_proto(self):
        """Test FileInfo wrapper conversion."""
        from mattermost_plugin._internal.wrappers import FileInfo
        from mattermost_plugin.grpc import file_pb2

        # Create a protobuf FileInfo
        proto_file_info = file_pb2.FileInfo(
            id="file123",
            creator_id="user123",
            post_id="post123",
            channel_id="channel123",
            create_at=1234567890000,
            name="document.pdf",
            extension="pdf",
            size=1024000,
            mime_type="application/pdf",
            width=0,
            height=0,
            has_preview_image=False,
        )

        # Convert to wrapper
        file_info = FileInfo.from_proto(proto_file_info)

        # Verify fields
        assert file_info.id == "file123"
        assert file_info.creator_id == "user123"
        assert file_info.post_id == "post123"
        assert file_info.name == "document.pdf"
        assert file_info.extension == "pdf"
        assert file_info.size == 1024000
        assert file_info.mime_type == "application/pdf"

    def test_upload_session_from_proto_and_to_proto(self):
        """Test UploadSession wrapper round-trip conversion."""
        from mattermost_plugin._internal.wrappers import UploadSession
        from mattermost_plugin.grpc import api_file_bot_pb2

        # Create a protobuf UploadSession
        proto_session = api_file_bot_pb2.UploadSession(
            id="session123",
            type="attachment",
            user_id="user123",
            channel_id="channel123",
            filename="large_file.zip",
            file_size=10485760,
            file_offset=5242880,
        )

        # Convert to wrapper
        session = UploadSession.from_proto(proto_session)

        # Verify fields
        assert session.id == "session123"
        assert session.type == "attachment"
        assert session.user_id == "user123"
        assert session.filename == "large_file.zip"
        assert session.file_size == 10485760
        assert session.file_offset == 5242880

        # Convert back to proto
        proto_back = session.to_proto()

        # Verify round-trip
        assert proto_back.id == proto_session.id
        assert proto_back.filename == proto_session.filename


class TestKVWrapperTypes:
    """Tests for KV Store related wrapper types."""

    def test_plugin_kv_set_options_to_proto(self):
        """Test PluginKVSetOptions wrapper conversion."""
        from mattermost_plugin._internal.wrappers import PluginKVSetOptions

        # Create options
        options = PluginKVSetOptions(
            atomic=True,
            old_value=b"old_data",
            expire_in_seconds=3600,
        )

        # Convert to proto
        proto = options.to_proto()

        # Verify
        assert proto.atomic is True
        assert proto.old_value == b"old_data"
        assert proto.expire_in_seconds == 3600


class TestClientPostMethods:
    """Tests for Post-related client methods."""

    @pytest.fixture
    def mock_client(self):
        """Create a mocked client for testing."""
        from mattermost_plugin.client import PluginAPIClient

        client = PluginAPIClient(target="localhost:50051")
        client._stub = MagicMock()
        client._channel = MagicMock()
        return client

    def test_create_post_success(self, mock_client):
        """Test create_post with successful response."""
        from mattermost_plugin._internal.wrappers import Post
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.CreatePostResponse()
        mock_response.post.CopyFrom(post_pb2.Post(
            id="newpost123",
            channel_id="channel123",
            user_id="user123",
            message="Hello!",
            create_at=1234567890000,
        ))
        mock_client._stub.CreatePost.return_value = mock_response

        # Create a post
        post_to_create = Post(id="", channel_id="channel123", message="Hello!")
        created_post = mock_client.create_post(post_to_create)

        # Verify
        assert created_post.id == "newpost123"
        assert created_post.channel_id == "channel123"
        assert created_post.message == "Hello!"

    def test_get_post_success(self, mock_client):
        """Test get_post with successful response."""
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.GetPostResponse()
        mock_response.post.CopyFrom(post_pb2.Post(
            id="post123",
            channel_id="channel123",
            message="Test message",
        ))
        mock_client._stub.GetPost.return_value = mock_response

        # Get the post
        post = mock_client.get_post("post123")

        # Verify
        assert post.id == "post123"
        assert post.message == "Test message"

    def test_get_post_not_found(self, mock_client):
        """Test get_post when post is not found."""
        from mattermost_plugin.grpc import api_channel_post_pb2, common_pb2
        from mattermost_plugin.exceptions import NotFoundError

        # Mock the response with error
        mock_response = api_channel_post_pb2.GetPostResponse()
        mock_response.error.CopyFrom(common_pb2.AppError(
            id="api.post.get.not_found.app_error",
            message="Post not found",
            status_code=404,
        ))
        mock_client._stub.GetPost.return_value = mock_response

        # Call the method and expect exception
        with pytest.raises(NotFoundError) as exc_info:
            mock_client.get_post("nonexistent")

        assert "Post not found" in str(exc_info.value)

    def test_update_post_success(self, mock_client):
        """Test update_post with successful response."""
        from mattermost_plugin._internal.wrappers import Post
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.UpdatePostResponse()
        mock_response.post.CopyFrom(post_pb2.Post(
            id="post123",
            channel_id="channel123",
            message="Updated message",
            edit_at=1234567890001,
        ))
        mock_client._stub.UpdatePost.return_value = mock_response

        # Update the post
        post_to_update = Post(id="post123", channel_id="channel123", message="Updated message")
        updated_post = mock_client.update_post(post_to_update)

        # Verify
        assert updated_post.message == "Updated message"
        assert updated_post.edit_at == 1234567890001

    def test_delete_post_success(self, mock_client):
        """Test delete_post with successful response."""
        from mattermost_plugin.grpc import api_channel_post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.DeletePostResponse()
        mock_client._stub.DeletePost.return_value = mock_response

        # Delete should not raise
        mock_client.delete_post("post123")

        # Verify the request was made
        mock_client._stub.DeletePost.assert_called_once()

    def test_send_ephemeral_post(self, mock_client):
        """Test send_ephemeral_post."""
        from mattermost_plugin._internal.wrappers import Post
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.SendEphemeralPostResponse()
        mock_response.post.CopyFrom(post_pb2.Post(
            id="ephemeral123",
            channel_id="channel123",
            message="Only you can see this!",
        ))
        mock_client._stub.SendEphemeralPost.return_value = mock_response

        # Send ephemeral post
        post = Post(id="", channel_id="channel123", message="Only you can see this!")
        result = mock_client.send_ephemeral_post("user123", post)

        # Verify
        assert result.id == "ephemeral123"
        assert result.message == "Only you can see this!"

    def test_add_reaction(self, mock_client):
        """Test add_reaction."""
        from mattermost_plugin._internal.wrappers import Reaction
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.AddReactionResponse()
        mock_response.reaction.CopyFrom(post_pb2.Reaction(
            user_id="user123",
            post_id="post123",
            emoji_name="thumbsup",
            create_at=1234567890000,
        ))
        mock_client._stub.AddReaction.return_value = mock_response

        # Add reaction
        reaction = Reaction(user_id="user123", post_id="post123", emoji_name="thumbsup")
        result = mock_client.add_reaction(reaction)

        # Verify
        assert result.emoji_name == "thumbsup"
        assert result.user_id == "user123"

    def test_get_reactions(self, mock_client):
        """Test get_reactions."""
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.GetReactionsResponse()
        mock_response.reactions.append(post_pb2.Reaction(
            user_id="user1",
            post_id="post123",
            emoji_name="thumbsup",
        ))
        mock_response.reactions.append(post_pb2.Reaction(
            user_id="user2",
            post_id="post123",
            emoji_name="heart",
        ))
        mock_client._stub.GetReactions.return_value = mock_response

        # Get reactions
        reactions = mock_client.get_reactions("post123")

        # Verify
        assert len(reactions) == 2
        assert reactions[0].emoji_name == "thumbsup"
        assert reactions[1].emoji_name == "heart"

    def test_get_post_thread(self, mock_client):
        """Test get_post_thread."""
        from mattermost_plugin.grpc import api_channel_post_pb2, post_pb2

        # Mock the response
        mock_response = api_channel_post_pb2.GetPostThreadResponse()
        mock_response.post_list.order.append("root")
        mock_response.post_list.order.append("reply1")
        mock_response.post_list.posts["root"].CopyFrom(post_pb2.Post(
            id="root",
            channel_id="ch1",
            message="Root post",
        ))
        mock_response.post_list.posts["reply1"].CopyFrom(post_pb2.Post(
            id="reply1",
            channel_id="ch1",
            message="Reply",
            root_id="root",
        ))
        mock_client._stub.GetPostThread.return_value = mock_response

        # Get thread
        post_list = mock_client.get_post_thread("root")

        # Verify
        assert len(post_list.order) == 2
        assert "root" in post_list.posts
        assert "reply1" in post_list.posts


class TestClientFileMethods:
    """Tests for File-related client methods."""

    @pytest.fixture
    def mock_client(self):
        """Create a mocked client for testing."""
        from mattermost_plugin.client import PluginAPIClient

        client = PluginAPIClient(target="localhost:50051")
        client._stub = MagicMock()
        client._channel = MagicMock()
        return client

    def test_get_file_info_success(self, mock_client):
        """Test get_file_info with successful response."""
        from mattermost_plugin.grpc import api_file_bot_pb2, file_pb2

        # Mock the response
        mock_response = api_file_bot_pb2.GetFileInfoResponse()
        mock_response.file_info.CopyFrom(file_pb2.FileInfo(
            id="file123",
            name="document.pdf",
            extension="pdf",
            size=1024000,
            mime_type="application/pdf",
        ))
        mock_client._stub.GetFileInfo.return_value = mock_response

        # Get file info
        file_info = mock_client.get_file_info("file123")

        # Verify
        assert file_info.id == "file123"
        assert file_info.name == "document.pdf"
        assert file_info.size == 1024000

    def test_get_file_info_not_found(self, mock_client):
        """Test get_file_info when file is not found."""
        from mattermost_plugin.grpc import api_file_bot_pb2, common_pb2
        from mattermost_plugin.exceptions import NotFoundError

        # Mock the response with error
        mock_response = api_file_bot_pb2.GetFileInfoResponse()
        mock_response.error.CopyFrom(common_pb2.AppError(
            id="api.file.get_info.not_found.app_error",
            message="File not found",
            status_code=404,
        ))
        mock_client._stub.GetFileInfo.return_value = mock_response

        # Call the method and expect exception
        with pytest.raises(NotFoundError):
            mock_client.get_file_info("nonexistent")

    def test_get_file_success(self, mock_client):
        """Test get_file with successful response."""
        from mattermost_plugin.grpc import api_file_bot_pb2

        # Mock the response
        mock_response = api_file_bot_pb2.GetFileResponse()
        mock_response.data = b"file content data"
        mock_client._stub.GetFile.return_value = mock_response

        # Get file
        data = mock_client.get_file("file123")

        # Verify
        assert data == b"file content data"

    def test_upload_file_success(self, mock_client):
        """Test upload_file with successful response."""
        from mattermost_plugin.grpc import api_file_bot_pb2, file_pb2

        # Mock the response
        mock_response = api_file_bot_pb2.UploadFileResponse()
        mock_response.file_info.CopyFrom(file_pb2.FileInfo(
            id="newfile123",
            name="test.txt",
            extension="txt",
            size=100,
            mime_type="text/plain",
        ))
        mock_client._stub.UploadFile.return_value = mock_response

        # Upload file
        file_info = mock_client.upload_file(b"test content", "channel123", "test.txt")

        # Verify
        assert file_info.id == "newfile123"
        assert file_info.name == "test.txt"

    def test_get_file_link_success(self, mock_client):
        """Test get_file_link with successful response."""
        from mattermost_plugin.grpc import api_file_bot_pb2

        # Mock the response
        mock_response = api_file_bot_pb2.GetFileLinkResponse()
        mock_response.link = "https://example.com/files/file123"
        mock_client._stub.GetFileLink.return_value = mock_response

        # Get file link
        link = mock_client.get_file_link("file123")

        # Verify
        assert link == "https://example.com/files/file123"

    def test_copy_file_infos_success(self, mock_client):
        """Test copy_file_infos with successful response."""
        from mattermost_plugin.grpc import api_file_bot_pb2

        # Mock the response
        mock_response = api_file_bot_pb2.CopyFileInfosResponse()
        mock_response.file_ids.append("newfile1")
        mock_response.file_ids.append("newfile2")
        mock_client._stub.CopyFileInfos.return_value = mock_response

        # Copy file infos
        new_ids = mock_client.copy_file_infos("user123", ["file1", "file2"])

        # Verify
        assert new_ids == ["newfile1", "newfile2"]


class TestClientKVStoreMethods:
    """Tests for KV Store related client methods."""

    @pytest.fixture
    def mock_client(self):
        """Create a mocked client for testing."""
        from mattermost_plugin.client import PluginAPIClient

        client = PluginAPIClient(target="localhost:50051")
        client._stub = MagicMock()
        client._channel = MagicMock()
        return client

    def test_kv_set_success(self, mock_client):
        """Test kv_set with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVSetResponse()
        mock_client._stub.KVSet.return_value = mock_response

        # Set should not raise
        mock_client.kv_set("mykey", b"myvalue")

        # Verify the request was made correctly
        mock_client._stub.KVSet.assert_called_once()
        call_args = mock_client._stub.KVSet.call_args
        assert call_args[0][0].key == "mykey"
        assert call_args[0][0].value == b"myvalue"

    def test_kv_get_success(self, mock_client):
        """Test kv_get with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVGetResponse()
        mock_response.value = b"stored_value"
        mock_client._stub.KVGet.return_value = mock_response

        # Get value
        value = mock_client.kv_get("mykey")

        # Verify
        assert value == b"stored_value"

    def test_kv_get_not_found(self, mock_client):
        """Test kv_get when key does not exist."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response with empty value
        mock_response = api_kv_config_pb2.KVGetResponse()
        mock_response.value = b""
        mock_client._stub.KVGet.return_value = mock_response

        # Get value
        value = mock_client.kv_get("nonexistent")

        # Verify
        assert value is None

    def test_kv_delete_success(self, mock_client):
        """Test kv_delete with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVDeleteResponse()
        mock_client._stub.KVDelete.return_value = mock_response

        # Delete should not raise
        mock_client.kv_delete("mykey")

        # Verify
        mock_client._stub.KVDelete.assert_called_once()

    def test_kv_delete_all_success(self, mock_client):
        """Test kv_delete_all with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVDeleteAllResponse()
        mock_client._stub.KVDeleteAll.return_value = mock_response

        # Delete all should not raise
        mock_client.kv_delete_all()

        # Verify
        mock_client._stub.KVDeleteAll.assert_called_once()

    def test_kv_list_success(self, mock_client):
        """Test kv_list with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVListResponse()
        mock_response.keys.append("key1")
        mock_response.keys.append("key2")
        mock_response.keys.append("key3")
        mock_client._stub.KVList.return_value = mock_response

        # List keys
        keys = mock_client.kv_list()

        # Verify
        assert keys == ["key1", "key2", "key3"]

    def test_kv_set_with_expiry_success(self, mock_client):
        """Test kv_set_with_expiry with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVSetWithExpiryResponse()
        mock_client._stub.KVSetWithExpiry.return_value = mock_response

        # Set with expiry should not raise
        mock_client.kv_set_with_expiry("cache_key", b"data", 300)

        # Verify the request
        call_args = mock_client._stub.KVSetWithExpiry.call_args
        assert call_args[0][0].key == "cache_key"
        assert call_args[0][0].expire_in_seconds == 300

    def test_kv_compare_and_set_success(self, mock_client):
        """Test kv_compare_and_set with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVCompareAndSetResponse()
        mock_response.success = True
        mock_client._stub.KVCompareAndSet.return_value = mock_response

        # Compare and set
        result = mock_client.kv_compare_and_set("mykey", b"old", b"new")

        # Verify
        assert result is True

    def test_kv_compare_and_set_conflict(self, mock_client):
        """Test kv_compare_and_set when value doesn't match."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVCompareAndSetResponse()
        mock_response.success = False
        mock_client._stub.KVCompareAndSet.return_value = mock_response

        # Compare and set with conflict
        result = mock_client.kv_compare_and_set("mykey", b"wrong_old", b"new")

        # Verify
        assert result is False

    def test_kv_compare_and_delete_success(self, mock_client):
        """Test kv_compare_and_delete with successful response."""
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVCompareAndDeleteResponse()
        mock_response.success = True
        mock_client._stub.KVCompareAndDelete.return_value = mock_response

        # Compare and delete
        result = mock_client.kv_compare_and_delete("mykey", b"expected_value")

        # Verify
        assert result is True

    def test_kv_set_with_options_success(self, mock_client):
        """Test kv_set_with_options with successful response."""
        from mattermost_plugin._internal.wrappers import PluginKVSetOptions
        from mattermost_plugin.grpc import api_kv_config_pb2

        # Mock the response
        mock_response = api_kv_config_pb2.KVSetWithOptionsResponse()
        mock_response.success = True
        mock_client._stub.KVSetWithOptions.return_value = mock_response

        # Set with options
        options = PluginKVSetOptions(
            atomic=True,
            old_value=b"old",
            expire_in_seconds=3600,
        )
        result = mock_client.kv_set_with_options("mykey", b"new", options)

        # Verify
        assert result is True


class TestErrorHandling:
    """Tests for error handling across Post/File/KV domains."""

    @pytest.fixture
    def mock_client(self):
        """Create a mocked client for testing."""
        from mattermost_plugin.client import PluginAPIClient

        client = PluginAPIClient(target="localhost:50051")
        client._stub = MagicMock()
        client._channel = MagicMock()
        return client

    def test_grpc_unavailable_error(self, mock_client):
        """Test that gRPC UNAVAILABLE maps to UnavailableError."""
        from mattermost_plugin.exceptions import UnavailableError

        # Create a mock that behaves like a gRPC error
        class MockGrpcError(grpc.RpcError):
            def code(self):
                return grpc.StatusCode.UNAVAILABLE

            def details(self):
                return "Service unavailable"

        mock_client._stub.GetPost.side_effect = MockGrpcError()

        # Call the method and expect exception
        with pytest.raises(UnavailableError):
            mock_client.get_post("post123")

    def test_validation_error_on_create_post(self, mock_client):
        """Test that 400 status code maps to ValidationError."""
        from mattermost_plugin._internal.wrappers import Post
        from mattermost_plugin.grpc import api_channel_post_pb2, common_pb2
        from mattermost_plugin.exceptions import ValidationError

        # Mock the response with 400 error
        mock_response = api_channel_post_pb2.CreatePostResponse()
        mock_response.error.CopyFrom(common_pb2.AppError(
            id="model.post.is_valid.msg.app_error",
            message="Message is required",
            status_code=400,
        ))
        mock_client._stub.CreatePost.return_value = mock_response

        # Call the method and expect ValidationError
        with pytest.raises(ValidationError) as exc_info:
            mock_client.create_post(Post(id="", channel_id="ch1", message=""))

        assert exc_info.value.status_code == 400