feat: add pin and unpin methods (#223)

botx/bots/mixins/requests/chats.py

@@ -9,9 +9,11 @@
 from botx.clients.methods.v3.chats.chat_list import ChatList
 from botx.clients.methods.v3.chats.create import Create
 from botx.clients.methods.v3.chats.info import Info
+from botx.clients.methods.v3.chats.pin_message import PinMessage
 from botx.clients.methods.v3.chats.remove_user import RemoveUser
 from botx.clients.methods.v3.chats.stealth_disable import StealthDisable
 from botx.clients.methods.v3.chats.stealth_set import StealthSet
+from botx.clients.methods.v3.chats.unpin_message import UnpinMessage
 from botx.models.chats import BotChatList, ChatFromSearch
 from botx.models.enums import ChatTypes
 from botx.models.messages.sending.credentials import SendingCredentials
@@ -188,3 +190,37 @@ async def add_admin_roles(
             AddAdminRole(group_chat_id=chat_id, user_huids=user_huids),
             credentials=credentials,
         )
+
+    async def pin_message(
+        self: BotXMethodCallProtocol,
+        credentials: SendingCredentials,
+        chat_id: UUID,
+        sync_id: UUID,
+    ) -> None:
+        """Pin message in chat.
+
+        Arguments:
+            credentials: credentials for making request.
+            chat_id: ID of chat where message should be pinned.
+            sync_id: ID of message that should be pinned.
+        """
+        await self.call_method(
+            PinMessage(chat_id=chat_id, sync_id=sync_id),
+            credentials=credentials,
+        )
+
+    async def unpin_message(
+        self: BotXMethodCallProtocol,
+        credentials: SendingCredentials,
+        chat_id: UUID,
+    ) -> None:
+        """Unpin message in chat.
+
+        Arguments:
+            credentials: credentials for making request.
+            chat_id: ID of chat where message should be unpinned.
+        """
+        await self.call_method(
+            UnpinMessage(chat_id=chat_id),
+            credentials=credentials,
+        )

botx/bots/mixins/requests/mixin.py

@@ -4,8 +4,8 @@
 
 from loguru import logger
 
-from botx.bots.mixins.requests import (  # noqa: WPS235
-    bots,
+from botx.bots.mixins.requests import bots  # noqa: WPS235
+from botx.bots.mixins.requests import (
     chats,
     command,
     events,

botx/clients/methods/errors/permissions.py

@@ -0,0 +1,51 @@
+"""Definition for "no permission" error."""
+from typing import NoReturn
+from uuid import UUID
+
+from pydantic import BaseModel
+
+from botx.clients.methods.base import APIErrorResponse, BotXMethod
+from botx.clients.types.http import HTTPResponse
+from botx.exceptions import BotXAPIError
+
+
+class NoPermissionError(BotXAPIError):
+    """Error for raising when there is no permission for operation."""
+
+    message_template = (
+        "Bot doesn't have permission for this operation in chat {group_chat_id}"
+    )
+
+    #: ID of chat that was requested.
+    group_chat_id: UUID
+
+
+class NoPermissionErrorData(BaseModel):
+    """Data for error when there is no permission for operation."""
+
+    #: ID of chat that was requested.
+    group_chat_id: UUID
+
+
+def handle_error(method: BotXMethod, response: HTTPResponse) -> NoReturn:
+    """Handle "no permission" error response.
+
+    Arguments:
+        method: method which was made before error.
+        response: HTTP response from BotX API.
+
+    Raises:
+        NoPermissionError: raised always.
+    """
+    parsed_response = APIErrorResponse[NoPermissionErrorData].parse_obj(
+        response.json_body,
+    )
+
+    error_data = parsed_response.error_data
+    raise NoPermissionError(
+        url=method.url,
+        method=method.http_method,
+        response_content=response.json_body,
+        status_content=response.status_code,
+        group_chat_id=error_data.group_chat_id,
+    )

botx/clients/methods/v3/chats/pin_message.py

@@ -0,0 +1,24 @@
+"""Method for pinning message in chat."""
+from http import HTTPStatus
+from uuid import UUID
+
+from botx.clients.methods.base import AuthorizedBotXMethod
+from botx.clients.methods.errors import chat_not_found, permissions
+
+
+class PinMessage(AuthorizedBotXMethod[str]):
+    """Method for pinning message in chat."""
+
+    __url__ = "/api/v3/botx/chats/pin_message"
+    __method__ = "POST"
+    __returning__ = str
+    __errors_handlers__ = {
+        HTTPStatus.NOT_FOUND: chat_not_found.handle_error,
+        HTTPStatus.FORBIDDEN: permissions.handle_error,
+    }
+
+    #: ID of chat where message should be pinned.
+    chat_id: UUID
+
+    #: ID of message that should be pinned.
+    sync_id: UUID

botx/clients/methods/v3/chats/unpin_message.py

@@ -0,0 +1,21 @@
+"""Method for unpinning message in chat."""
+from http import HTTPStatus
+from uuid import UUID
+
+from botx.clients.methods.base import AuthorizedBotXMethod
+from botx.clients.methods.errors import chat_not_found, permissions
+
+
+class UnpinMessage(AuthorizedBotXMethod[str]):
+    """Method for unpinning message in chat."""
+
+    __url__ = "/api/v3/botx/chats/unpin_message"
+    __method__ = "POST"
+    __returning__ = str
+    __errors_handlers__ = {
+        HTTPStatus.NOT_FOUND: chat_not_found.handle_error,
+        HTTPStatus.FORBIDDEN: permissions.handle_error,
+    }
+
+    #: ID of chat where message should be unpinned.
+    chat_id: UUID

botx/testing/botx_mock/asgi/application.py

@@ -8,8 +8,8 @@
 
 from botx.clients.methods.base import BotXMethod
 from botx.testing.botx_mock.asgi.errors import ErrorMiddleware
-from botx.testing.botx_mock.asgi.routes import (  # noqa: WPS235
-    bots,
+from botx.testing.botx_mock.asgi.routes import bots  # noqa: WPS235
+from botx.testing.botx_mock.asgi.routes import (
     chats,
     command,
     events,
@@ -35,6 +35,8 @@
     chats.post_stealth_set,
     chats.post_stealth_disable,
     chats.post_create,
+    chats.post_pin_message,
+    chats.post_unpin_message,
     # command
     command.post_command_result,
     # events

botx/testing/botx_mock/asgi/routes/chats.py

@@ -5,15 +5,17 @@
 from starlette import requests, responses
 
 from botx.clients.methods.base import APIResponse
-from botx.clients.methods.v3.chats import (
+from botx.clients.methods.v3.chats import (  # noqa: WPS235
     add_admin_role,
     add_user,
     chat_list,
     create,
     info,
+    pin_message,
     remove_user,
     stealth_disable,
     stealth_set,
+    unpin_message,
 )
 from botx.clients.types.response_results import ChatCreatedResult
 from botx.models import chats, enums, users
@@ -180,3 +182,33 @@ async def post_add_admin_role(request: requests.Request) -> responses.Response:
     payload = add_admin_role.AddAdminRole.parse_obj(await request.json())
     add_request_to_collection(request, payload)
     return PydanticResponse(APIResponse[bool](result=True))
+
+
+@bind_implementation_to_method(pin_message.PinMessage)
+async def post_pin_message(request: requests.Request) -> responses.Response:
+    """Handle pinning message in chat request.
+
+    Arguments:
+        request: HTTP request from Starlette.
+
+    Returns:
+        Response with result of pinning.
+    """
+    payload = pin_message.PinMessage.parse_obj(await request.json())
+    add_request_to_collection(request, payload)
+    return PydanticResponse(APIResponse[str](result="pinned"))
+
+
+@bind_implementation_to_method(unpin_message.UnpinMessage)
+async def post_unpin_message(request: requests.Request) -> responses.Response:
+    """Handle unpinning message in chat request.
+
+    Arguments:
+        request: HTTP request from Starlette.
+
+    Returns:
+        Response with result of unpinning.
+    """
+    payload = unpin_message.UnpinMessage.parse_obj(await request.json())
+    add_request_to_collection(request, payload)
+    return PydanticResponse(APIResponse[str](result="unpinned"))

botx/testing/botx_mock/wsgi/application.py

@@ -6,8 +6,8 @@
 
 from botx.clients.methods.base import BotXMethod
 from botx.testing.botx_mock.wsgi.errors import error_middleware
-from botx.testing.botx_mock.wsgi.routes import (  # noqa: WPS235
-    bots,
+from botx.testing.botx_mock.wsgi.routes import bots  # noqa: WPS235
+from botx.testing.botx_mock.wsgi.routes import (
     chats,
     command,
     events,
@@ -33,6 +33,8 @@
     chats.post_stealth_set,
     chats.post_stealth_disable,
     chats.post_create,
+    chats.post_pin_message,
+    chats.post_unpin_message,
     # command
     command.post_command_result,
     # events

botx/testing/botx_mock/wsgi/routes/chats.py

@@ -5,15 +5,17 @@
 from molten import Request, RequestData, Response, Settings
 
 from botx.clients.methods.base import APIResponse
-from botx.clients.methods.v3.chats import (
+from botx.clients.methods.v3.chats import (  # noqa: WPS235
     add_admin_role,
     add_user,
     chat_list,
     create,
     info,
+    pin_message,
     remove_user,
     stealth_disable,
     stealth_set,
+    unpin_message,
 )
 from botx.clients.types.response_results import ChatCreatedResult
 from botx.models import chats, enums, users
@@ -188,3 +190,35 @@ def post_add_admin_role(request_data: RequestData, settings: Settings) -> Respon
     payload = add_admin_role.AddAdminRole.parse_obj(request_data)
     add_request_to_collection(settings, payload)
     return PydanticResponse(APIResponse[bool](result=True))
+
+
+@bind_implementation_to_method(pin_message.PinMessage)
+def post_pin_message(request_data: RequestData, settings: Settings) -> Response:
+    """Handle pinning message in chat request.
+
+    Arguments:
+        request_data: parsed json data from request.
+        settings: application settings with storage.
+
+    Returns:
+        Response with result of pinning.
+    """
+    payload = pin_message.PinMessage.parse_obj(request_data)
+    add_request_to_collection(settings, payload)
+    return PydanticResponse(APIResponse[str](result="pinned"))
+
+
+@bind_implementation_to_method(unpin_message.UnpinMessage)
+def post_unpin_message(request_data: RequestData, settings: Settings) -> Response:
+    """Handle unpinning message in chat request.
+
+    Arguments:
+        request_data: parsed json data from request.
+        settings: application settings with storage.
+
+    Returns:
+        Response with result of unpinning.
+    """
+    payload = unpin_message.UnpinMessage.parse_obj(request_data)
+    add_request_to_collection(settings, payload)
+    return PydanticResponse(APIResponse[str](result="pinned"))

docs/changelog.md

@@ -1,3 +1,10 @@
+## 0.27.0 (Nov 8, 2021)
+
+### Added
+
+* `pin_message` and `unpin_message` methods.
+
+
 ## 0.26.0 (Nov 1, 2021)
 
 ### Added

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "botx"
-version = "0.26.0"
+version = "0.27.0"
 description = "A little python framework for building bots for eXpress"
 license = "MIT"
 authors = [

tests/test_bots/test_mixins/test_requests/test_chats.py

@@ -92,3 +92,30 @@ async def test_promoting_users_to_admins(bot, client, message):
     assert request.group_chat_id == message.group_chat_id
 
     assert request.user_huids == users
+
+
+async def test_pinning_message(bot, client, message):
+    chat_id = uuid.uuid4()
+    sync_id = uuid.uuid4()
+
+    await bot.pin_message(
+        message.credentials,
+        chat_id=chat_id,
+        sync_id=sync_id,
+    )
+    request = client.requests[0]
+
+    assert request.chat_id == chat_id
+    assert request.sync_id == sync_id
+
+
+async def test_unpinning_message(bot, client, message):
+    chat_id = uuid.uuid4()
+
+    await bot.unpin_message(
+        message.credentials,
+        chat_id=chat_id,
+    )
+    request = client.requests[0]
+
+    assert request.chat_id == chat_id

tests/test_clients/test_methods/test_errors/test_permissions.py

@@ -0,0 +1,40 @@
+import uuid
+from http import HTTPStatus
+
+import pytest
+
+from botx.clients.methods.errors.permissions import (
+    NoPermissionError,
+    NoPermissionErrorData,
+)
+from botx.clients.methods.v3.chats.pin_message import PinMessage
+from botx.concurrency import callable_to_coroutine
+
+pytestmark = pytest.mark.asyncio
+pytest_plugins = ("tests.test_clients.fixtures",)
+
+
+async def test_raising_no_permission(client, requests_client):
+    method = PinMessage(
+        host="example.com",
+        chat_id=uuid.uuid4(),
+        sync_id=uuid.uuid4(),
+    )
+
+    errors_to_raise = {
+        PinMessage: (
+            HTTPStatus.FORBIDDEN,
+            NoPermissionErrorData(group_chat_id=method.chat_id),
+        ),
+    }
+
+    with client.error_client(errors=errors_to_raise):
+        request = requests_client.build_request(method)
+        response = await callable_to_coroutine(requests_client.execute, request)
+
+        with pytest.raises(NoPermissionError):
+            await callable_to_coroutine(
+                requests_client.process_response,
+                method,
+                response,
+            )