fix: Add handling of unknown type attachment. (#337)

* fix: Add handling of unknown type attachment.

* chore: Bump package version.

botx/models/message/incoming_message.py

@@ -5,6 +5,7 @@
 
 from pydantic import Field
 
+from botx.logger import logger
 from botx.models.async_files import APIAsyncFile, File, convert_async_file_to_domain
 from botx.models.attachments import (
     AttachmentContact,
@@ -175,7 +176,7 @@ class BotAPIIncomingMessage(BotAPIBaseCommand):
     sender: BotAPIIncomingMessageContext = Field(..., alias="from")
 
     source_sync_id: Optional[UUID]
-    attachments: List[BotAPIAttachment]
+    attachments: List[Union[BotAPIAttachment, Dict[str, Any]]]  # noqa: WPS234
     async_files: List[APIAsyncFile]
     entities: List[BotAPIEntity]
 
@@ -223,22 +224,28 @@ def to_domain(self, raw_command: Dict[str, Any]) -> IncomingMessage:  # noqa: WP
         location: Optional[AttachmentLocation] = None
         contact: Optional[AttachmentContact] = None
         link: Optional[AttachmentLink] = None
+
         if self.async_files:
             # Always one async file per-message
             file = convert_async_file_to_domain(self.async_files[0])
         elif self.attachments:
             # Always one attachment per-message
-            attachment_domain = convert_api_attachment_to_domain(self.attachments[0])
-            if isinstance(attachment_domain, FileAttachmentBase):
-                file = attachment_domain
-            elif attachment_domain.type == AttachmentTypes.LOCATION:
-                location = attachment_domain
-            elif attachment_domain.type == AttachmentTypes.CONTACT:
-                contact = attachment_domain
-            elif attachment_domain.type == AttachmentTypes.LINK:
-                link = attachment_domain
+            if isinstance(self.attachments[0], dict):
+                logger.warning("Received unknown attachment type")
             else:
-                raise NotImplementedError
+                attachment_domain = convert_api_attachment_to_domain(
+                    self.attachments[0],
+                )
+                if isinstance(attachment_domain, FileAttachmentBase):
+                    file = attachment_domain
+                elif attachment_domain.type == AttachmentTypes.LOCATION:
+                    location = attachment_domain
+                elif attachment_domain.type == AttachmentTypes.CONTACT:
+                    contact = attachment_domain
+                elif attachment_domain.type == AttachmentTypes.LINK:
+                    link = attachment_domain
+                else:
+                    raise NotImplementedError
 
         mentions: MentionList = MentionList()
         forward: Optional[Forward] = None

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "botx"
-version = "0.30.2"
+version = "0.30.3"
 description = "A python library for interacting with eXpress BotX API"
 authors = [
     "Sidnev Nikolay <[email protected]>",

tests/test_attachments.py

@@ -270,3 +270,24 @@ async def default_handler(message: IncomingMessage, bot: Bot) -> None:
     # - Assert -
     assert incoming_message
     assert incoming_message.file == domain_attachment
+
+
+async def test__async_execute_raw_bot_command__unknown_attachment_type(
+    api_incoming_message_factory: Callable[..., Dict[str, Any]],
+    bot_account: BotAccountWithSecret,
+    loguru_caplog: pytest.LogCaptureFixture,
+) -> None:
+    # - Arrange -
+    unknown_attachment = {"data": {"foo": "bar"}, "type": "baz"}
+    payload = api_incoming_message_factory(attachment=unknown_attachment)
+
+    collector = HandlerCollector()
+
+    built_bot = Bot(collectors=[collector], bot_accounts=[bot_account])
+
+    # - Act -
+    async with lifespan_wrapper(built_bot) as bot:
+        bot.async_execute_raw_bot_command(payload)
+
+    # - Assert -
+    assert "Received unknown attachment type" in loguru_caplog.text