diff --git a/tests/data/18.0/message.json b/tests/data/18.0/message.json index 9670226b..1d022aa7 100644 --- a/tests/data/18.0/message.json +++ b/tests/data/18.0/message.json @@ -381,6 +381,46 @@ } ] }, + "unreaction": { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "1234567890987654321", + "changes": [ + { + "value": { + "messaging_product": "whatsapp", + "metadata": { + "display_phone_number": "972123456789", + "phone_number_id": "1122334455667" + }, + "contacts": [ + { + "profile": { + "name": "Test Name" + }, + "wa_id": "972987654321" + } + ], + "messages": [ + { + "from": "972987654321", + "id": "wamid.xyzxyz", + "timestamp": "1697044456", + "type": "reaction", + "reaction": { + "message_id": "wamid.yzxyzx=", + "emoji": "" + } + } + ] + }, + "field": "messages" + } + ] + } + ] + }, "location": { "object": "whatsapp_business_account", "entry": [ @@ -632,6 +672,10 @@ ], "messages": [ { + "context": { + "from": "972123456789", + "id": "wamid.xyzxyz==" + }, "from": "972987654321", "id": "wamid.xyzxyz", "timestamp": "1697044746", diff --git a/tests/test_updates.py b/tests/test_updates.py index 146befc4..bb4e714a 100644 --- a/tests/test_updates.py +++ b/tests/test_updates.py @@ -1,5 +1,5 @@ import json -from typing import Any +from typing import Any, TypeVar, Callable from pywa import WhatsApp from pywa.types import ( @@ -8,59 +8,88 @@ CallbackSelection, MessageStatus, TemplateStatus, + MessageType, + MessageStatusType, ) from pywa.types.base_update import BaseUpdate +from pywa.types.media import Image, Video, Document, Audio API_VERSIONS: list[float] = [18.0] +T = TypeVar("T", bound=BaseUpdate) -UPDATES: dict[tuple[str, type[BaseUpdate]], list[str]] = { - ("message", Message): [ - "text", - "image", - "video", - "document", - "audio", - "voice", - "static_sticker", - "animated_sticker", - "location", - "contacts", - "order", - "unsupported", - "reply", - "forwarded", - "forwarded_many_times", - ], - ("callback_button", CallbackButton): [ - "button", - "quick_reply", - ], - ("callback_selection", CallbackSelection): [ - "callback", - "description", - ], - ("message_status", MessageStatus): [ - "sent", - "failed", - ], - ("template_status", TemplateStatus): [ - "approved", - ], +UPDATES: dict[tuple[str, type[T]], dict[str, list[Callable[[T], bool]]]] = { + ("message", Message): { + "text": [lambda m: m.text is not None], + "image": [ + lambda m: m.image is not None, + lambda m: m.has_media, + lambda m: isinstance(m.media, Image), + ], + "video": [ + lambda m: m.video is not None, + lambda m: m.has_media, + lambda m: isinstance(m.media, Video), + ], + "document": [ + lambda m: m.document is not None, + lambda m: m.has_media, + lambda m: isinstance(m.media, Document), + ], + "audio": [ + lambda m: not m.audio.voice, + lambda m: m.has_media, + lambda m: isinstance(m.media, Audio), + ], + "voice": [lambda m: m.audio.voice], + "static_sticker": [lambda m: not m.sticker.animated], + "animated_sticker": [lambda m: m.sticker.animated], + "reaction": [ + lambda m: m.reaction.emoji is not None, + lambda m: m.id != m.message_id_to_reply, + ], + "unreaction": [lambda m: m.reaction.emoji is None], + "location": [lambda m: m.location is not None], + "contacts": [lambda m: m.contacts is not None], + "order": [lambda m: m.order is not None], + "unsupported": [lambda m: m.error is not None], + "reply": [lambda m: m.is_reply], + "forwarded": [lambda m: m.forwarded], + "forwarded_many_times": [lambda m: m.forwarded_many_times], + }, + ("callback_button", CallbackButton): { + "button": [lambda b: b.type == MessageType.INTERACTIVE], + "quick_reply": [lambda b: b.type == MessageType.BUTTON], + }, + ("callback_selection", CallbackSelection): { + "callback": [lambda s: s.data is not None], + "description": [lambda s: s.description is not None], + }, + ("message_status", MessageStatus): { + "sent": [lambda s: s.status == MessageStatusType.SENT], + "failed": [lambda s: s.error is not None], + }, + ("template_status", TemplateStatus): { + "approved": [lambda s: s.event == TemplateStatus.TemplateEvent.APPROVED], + }, } def test_types(): client = WhatsApp(phone_id="1234567890", token="xyzxyzxyz") for version in API_VERSIONS: - for (update_id, update_type), updates in UPDATES.items(): - with open(f"tests/data/{version}/{update_id}.json", "r") as update_file: - update_data: dict[str, Any] = json.load(update_file) - for update in updates: - try: - update = update_type.from_update( - client=client, update=update_data[update] - ) # noqa - except Exception as e: - raise AssertionError( - f"Failed to parse update '{update_id}': test='{update}', v={version}, error={e}" - ) + for (update_filename, update_class), examples in UPDATES.items(): + with open( + f"tests/data/{version}/{update_filename}.json", "r" + ) as update_file: + examples_data: dict[str, Any] = json.load(update_file) + for test_name, test_funcs in examples.items(): + try: + update_obj = update_class.from_update( + client=client, update=examples_data[test_name] + ) # noqa + for test_func in test_funcs: + assert test_func(update_obj) + except Exception as e: + raise AssertionError( + f"Failed to parse update='{update_filename}', test='{test_name}', v={version}, error={e}" + )