[webhooks] initial changes, WIP

Author: capcom6

Makefile

@@ -6,7 +6,7 @@ VERSION=$(shell grep '__version__' $(PACKAGE_NAME)/__init__.py | cut -d '"' -f 2
 
 # Install pipenv and project dependencies
 install:
-	pipenv install --dev
+	pipenv install --dev --categories encryption
 
 # Run tests with pytest or unittest
 test:

android_sms_gateway/client.py

@@ -114,6 +114,26 @@ def get_state(self, _id: str) -> domain.MessageState:
             )
         )
 
+    def get_webhooks(self) -> t.List[domain.Webhook]:
+        return [
+            domain.Webhook.from_dict(webhook)
+            for webhook in self.http.get(
+                f"{self.base_url}/webhooks", headers=self.headers
+            )
+        ]
+
+    def create_webhook(self, webhook: domain.Webhook) -> domain.Webhook:
+        return domain.Webhook.from_dict(
+            self.http.post(
+                f"{self.base_url}/webhooks",
+                payload=webhook.asdict(),
+                headers=self.headers,
+            )
+        )
+
+    def delete_webhook(self, _id: str) -> None:
+        self.http.delete(f"{self.base_url}/webhooks/{_id}", headers=self.headers)
+
 
 class AsyncAPIClient(BaseClient):
     def __init__(

android_sms_gateway/domain.py

@@ -1,7 +1,7 @@
 import dataclasses
 import typing as t
 
-from .enums import ProcessState
+from .enums import ProcessState, WebhookEvent
 
 
 def snake_to_camel(snake_str):
@@ -63,3 +63,43 @@ def from_dict(cls, payload: t.Dict[str, t.Any]) -> "MessageState":
             is_hashed=payload.get("isHashed", False),
             is_encrypted=payload.get("isEncrypted", False),
         )
+
+
+@dataclasses.dataclass(frozen=True)
+class Webhook:
+    """A webhook configuration."""
+
+    id: t.Optional[str]
+    """The unique identifier of the webhook."""
+    url: str
+    """The URL the webhook will be sent to."""
+    event: WebhookEvent
+    """The type of event the webhook is triggered for."""
+
+    @classmethod
+    def from_dict(cls, payload: t.Dict[str, t.Any]) -> "Webhook":
+        """Creates a Webhook instance from a dictionary.
+
+        Args:
+            payload: A dictionary containing the webhook's data.
+
+        Returns:
+            A Webhook instance.
+        """
+        return cls(
+            id=payload["id"],
+            url=payload["url"],
+            event=WebhookEvent(payload["event"]),
+        )
+
+    def asdict(self) -> t.Dict[str, t.Any]:
+        """Returns a dictionary representation of the webhook.
+
+        Returns:
+            A dictionary containing the webhook's data.
+        """
+        return {
+            "id": self.id,
+            "url": self.url,
+            "event": self.event.value,
+        }

android_sms_gateway/enums.py

@@ -7,3 +7,24 @@ class ProcessState(enum.Enum):
     Sent = "Sent"
     Delivered = "Delivered"
     Failed = "Failed"
+
+
+class WebhookEvent(enum.Enum):
+    """
+    Webhook events that can be sent by the server.
+    """
+
+    SMS_RECEIVED = "sms:received"
+    """Triggered when an SMS is received."""
+
+    SMS_SENT = "sms:sent"
+    """Triggered when an SMS is sent."""
+
+    SMS_DELIVERED = "sms:delivered"
+    """Triggered when an SMS is delivered."""
+
+    SMS_FAILED = "sms:failed"
+    """Triggered when an SMS processing fails."""
+
+    SYSTEM_PING = "system:ping"
+    """Triggered when the device pings the server."""

android_sms_gateway/http.py

@@ -13,6 +13,19 @@ def post(
         self, url: str, payload: dict, *, headers: t.Optional[t.Dict[str, str]] = None
     ) -> dict: ...
 
+    @abc.abstractmethod
+    def delete(self, url: str, *, headers: t.Optional[t.Dict[str, str]] = None) -> None:
+        """
+        Sends a DELETE request to the specified URL.
+
+        Args:
+            url: The URL to send the DELETE request to.
+            headers: Optional dictionary of HTTP headers to send with the request.
+
+        Returns:
+            None
+        """
+
     def __enter__(self):
         pass
 

pyproject.toml

@@ -42,3 +42,4 @@ dev = ["setuptools", "pytest", "black", "flake8", "wheel"]
 requests = ["requests"]
 httpx = ["httpx"]
 aiohttp = ["aiohttp"]
+encryption = ["pycryptodome"]

tests/test_domain.py

@@ -1,6 +1,7 @@
+from android_sms_gateway.enums import WebhookEvent
 import pytest
 
-from android_sms_gateway.domain import MessageState, RecipientState
+from android_sms_gateway.domain import MessageState, RecipientState, Webhook
 
 
 # Test for successful instantiation from a dictionary
@@ -79,3 +80,58 @@ def test_message_state_from_dict_incorrect_types():
         Exception
     ):  # Replace Exception with the specific exception you expect
         MessageState.from_dict(incorrect_payload)
+
+
+def test_webhook_from_dict():
+    """
+    Tests that a Webhook instance can be successfully instantiated from a dictionary
+    representation of a webhook.
+    """
+    payload = {
+        "id": "webhook_123",
+        "url": "https://example.com/webhook",
+        "event": "sms:received",
+    }
+
+    webhook = Webhook.from_dict(payload)
+
+    assert webhook.id == payload["id"]
+    assert webhook.url == payload["url"]
+    assert webhook.event == WebhookEvent(payload["event"])
+
+
+def test_webhook_asdict():
+    """
+    Tests that a Webhook instance can be successfully converted to a dictionary
+    representation and that the fields match the expected values.
+
+    This test ensures that the asdict method of the Webhook class returns a dictionary
+    with the correct keys and values.
+    """
+    webhook = Webhook(
+        id="webhook_123",
+        url="https://example.com/webhook",
+        event=WebhookEvent.SMS_RECEIVED,
+    )
+
+    expected_dict = {
+        "id": "webhook_123",
+        "url": "https://example.com/webhook",
+        "event": "sms:received",
+    }
+
+    assert webhook.asdict() == expected_dict
+
+    webhook = Webhook(
+        id=None,
+        url="https://example.com/webhook",
+        event=WebhookEvent.SMS_RECEIVED,
+    )
+
+    expected_dict = {
+        "id": None,
+        "url": "https://example.com/webhook",
+        "event": "sms:received",
+    }
+
+    assert webhook.asdict() == expected_dict