Fix issue #23: Add SuppressionsApi, related models, tests and examples

Author: Ihor Bilous

examples/suppressions.py

@@ -0,0 +1,22 @@
+from typing import Optional
+
+import mailtrap as mt
+from mailtrap.models.suppressions import Suppression
+
+API_TOKEN = "YOU_API_TOKEN"
+ACCOUNT_ID = "YOU_ACCOUNT_ID"
+
+client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID)
+suppressions_api = client.suppressions_api.suppressions
+
+
+def list_suppressions(email: Optional[str] = None) -> list[Suppression]:
+    return suppressions_api.get_list(email)
+
+
+def delete_suppression(suppression_id: str) -> Suppression:
+    return suppressions_api.delete(suppression_id)
+
+
+if __name__ == "__main__":
+    print(list_suppressions())

mailtrap/api/resources/suppressions.py

@@ -0,0 +1,25 @@
+from typing import Optional
+
+from mailtrap.http import HttpClient
+from mailtrap.models.suppressions import Suppression
+
+
+class SuppressionsApi:
+    def __init__(self, client: HttpClient, account_id: str) -> None:
+        self._account_id = account_id
+        self._client = client
+
+    def get_list(self, email: Optional[str] = None) -> list[Suppression]:
+        params = {"email": email} if email is not None else None
+        response = self._client.get(self._api_path(), params=params)
+        return [Suppression(**suppression) for suppression in response]
+
+    def delete(self, suppression_id: str) -> Suppression:
+        response = self._client.delete(self._api_path(suppression_id))
+        return Suppression(**response)
+
+    def _api_path(self, suppression_id: Optional[str] = None) -> str:
+        path = f"/api/accounts/{self._account_id}/suppressions"
+        if suppression_id is not None:
+            return f"{path}/{suppression_id}"
+        return path

mailtrap/api/suppressions.py

@@ -0,0 +1,12 @@
+from mailtrap.api.resources.suppressions import SuppressionsApi
+from mailtrap.http import HttpClient
+
+
+class SuppressionsBaseApi:
+    def __init__(self, client: HttpClient, account_id: str) -> None:
+        self._account_id = account_id
+        self._client = client
+
+    @property
+    def suppressions(self) -> SuppressionsApi:
+        return SuppressionsApi(account_id=self._account_id, client=self._client)

mailtrap/client.py

@@ -7,6 +7,7 @@
 
 from mailtrap.api.contacts import ContactsBaseApi
 from mailtrap.api.sending import SendingApi
+from mailtrap.api.suppressions import SuppressionsBaseApi
 from mailtrap.api.templates import EmailTemplatesApi
 from mailtrap.api.testing import TestingApi
 from mailtrap.config import BULK_HOST
@@ -72,6 +73,14 @@ def contacts_api(self) -> ContactsBaseApi:
             client=HttpClient(host=GENERAL_HOST, headers=self.headers),
         )
 
+    @property
+    def suppressions_api(self) -> SuppressionsBaseApi:
+        self._validate_account_id()
+        return SuppressionsBaseApi(
+            account_id=cast(str, self.account_id),
+            client=HttpClient(host=GENERAL_HOST, headers=self.headers),
+        )
+
     @property
     def sending_api(self) -> SendingApi:
         http_client = HttpClient(host=self._sending_api_host, headers=self.headers)

mailtrap/models/suppressions.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+from typing import Union
+
+from pydantic.dataclasses import dataclass
+
+
+class SuppressionType(str, Enum):
+    HARD_BOUNCE = "hard bounce"
+    SPAM_COMPLAINT = "spam complaint"
+    UNSUBSCRIPTION = "unsubscription"
+    MANUAL_IMPORT = "manual import"
+
+
+class SendingStream(str, Enum):
+    TRANSACTIONAL = "transactional"
+    BULK = "bulk"
+
+
+@dataclass
+class Suppression:
+    id: str
+    type: SuppressionType
+    created_at: datetime
+    email: str
+    sending_stream: SendingStream
+    domain_name: Optional[str] = None
+    message_bounce_category: Optional[str] = None
+    message_category: Optional[str] = None
+    message_client_ip: Optional[str] = None
+    message_created_at: Optional[Union[str, datetime]] = None
+    message_esp_response: Optional[str] = None
+    message_esp_server_type: Optional[str] = None
+    message_outgoing_ip: Optional[str] = None
+    message_recipient_mx_name: Optional[str] = None
+    message_sender_email: Optional[str] = None
+    message_subject: Optional[str] = None

tests/unit/api/test_suppressions.py

@@ -0,0 +1,173 @@
+from datetime import datetime
+from typing import Any
+
+import pytest
+import responses
+
+from mailtrap.api.resources.suppressions import SuppressionsApi
+from mailtrap.config import GENERAL_HOST
+from mailtrap.exceptions import APIError
+from mailtrap.http import HttpClient
+from mailtrap.models.suppressions import SendingStream
+from mailtrap.models.suppressions import Suppression
+from mailtrap.models.suppressions import SuppressionType
+from tests import conftest
+
+ACCOUNT_ID = "321"
+SUPPRESSION_ID = "supp_123456"
+BASE_SUPPRESSIONS_URL = f"https://{GENERAL_HOST}/api/accounts/{ACCOUNT_ID}/suppressions"
+
+
+@pytest.fixture
+def suppressions_api() -> SuppressionsApi:
+    return SuppressionsApi(account_id=ACCOUNT_ID, client=HttpClient(GENERAL_HOST))
+
+
+@pytest.fixture
+def sample_suppression_dict() -> dict[str, Any]:
+    return {
+        "id": SUPPRESSION_ID,
+        "type": "unsubscription",
+        "created_at": "2024-12-26T09:40:44.161Z",
+        "email": "[email protected]",
+        "sending_stream": "transactional",
+        "domain_name": "sender.com",
+        "message_bounce_category": None,
+        "message_category": "Welcome email",
+        "message_client_ip": "123.123.123.123",
+        "message_created_at": "2024-12-26T07:10:00.889Z",
+        "message_outgoing_ip": "1.1.1.1",
+        "message_recipient_mx_name": "Other Providers",
+        "message_sender_email": "[email protected]",
+        "message_subject": "Welcome!",
+    }
+
+
+class TestSuppressionsApi:
+
+    @pytest.mark.parametrize(
+        "status_code,response_json,expected_error_message",
+        [
+            (
+                conftest.UNAUTHORIZED_STATUS_CODE,
+                conftest.UNAUTHORIZED_RESPONSE,
+                conftest.UNAUTHORIZED_ERROR_MESSAGE,
+            ),
+            (
+                conftest.FORBIDDEN_STATUS_CODE,
+                conftest.FORBIDDEN_RESPONSE,
+                conftest.FORBIDDEN_ERROR_MESSAGE,
+            ),
+        ],
+    )
+    @responses.activate
+    def test_get_suppressions_should_raise_api_errors(
+        self,
+        suppressions_api: SuppressionsApi,
+        status_code: int,
+        response_json: dict,
+        expected_error_message: str,
+    ) -> None:
+        responses.get(
+            BASE_SUPPRESSIONS_URL,
+            status=status_code,
+            json=response_json,
+        )
+
+        with pytest.raises(APIError) as exc_info:
+            suppressions_api.get_list()
+
+        assert expected_error_message in str(exc_info.value)
+
+    @responses.activate
+    def test_get_suppressions_should_return_suppression_list(
+        self, suppressions_api: SuppressionsApi, sample_suppression_dict: dict
+    ) -> None:
+        responses.get(
+            BASE_SUPPRESSIONS_URL,
+            json=[sample_suppression_dict],
+            status=200,
+        )
+
+        suppressions = suppressions_api.get_list()
+
+        assert isinstance(suppressions, list)
+        assert all(isinstance(s, Suppression) for s in suppressions)
+        assert suppressions[0].id == SUPPRESSION_ID
+
+    @responses.activate
+    def test_get_suppressions_with_email_filter_should_return_filtered_list(
+        self, suppressions_api: SuppressionsApi, sample_suppression_dict: dict
+    ) -> None:
+        email_filter = "[email protected]"
+        responses.get(
+            BASE_SUPPRESSIONS_URL,
+            json=[sample_suppression_dict],
+            status=200,
+            match=[responses.matchers.query_param_matcher({"email": email_filter})],
+        )
+
+        suppressions = suppressions_api.get_list(email=email_filter)
+
+        assert isinstance(suppressions, list)
+        assert all(isinstance(s, Suppression) for s in suppressions)
+        assert suppressions[0].id == SUPPRESSION_ID
+        assert suppressions[0].email == email_filter
+        assert isinstance(suppressions[0].created_at, datetime)
+
+    @pytest.mark.parametrize(
+        "status_code,response_json,expected_error_message",
+        [
+            (
+                conftest.UNAUTHORIZED_STATUS_CODE,
+                conftest.UNAUTHORIZED_RESPONSE,
+                conftest.UNAUTHORIZED_ERROR_MESSAGE,
+            ),
+            (
+                conftest.FORBIDDEN_STATUS_CODE,
+                conftest.FORBIDDEN_RESPONSE,
+                conftest.FORBIDDEN_ERROR_MESSAGE,
+            ),
+            (
+                conftest.NOT_FOUND_STATUS_CODE,
+                conftest.NOT_FOUND_RESPONSE,
+                conftest.NOT_FOUND_ERROR_MESSAGE,
+            ),
+        ],
+    )
+    @responses.activate
+    def test_delete_suppression_should_raise_api_errors(
+        self,
+        suppressions_api: SuppressionsApi,
+        status_code: int,
+        response_json: dict,
+        expected_error_message: str,
+    ) -> None:
+        responses.delete(
+            f"{BASE_SUPPRESSIONS_URL}/{SUPPRESSION_ID}",
+            status=status_code,
+            json=response_json,
+        )
+
+        with pytest.raises(APIError) as exc_info:
+            suppressions_api.delete(SUPPRESSION_ID)
+
+        assert expected_error_message in str(exc_info.value)
+
+    @responses.activate
+    def test_delete_suppression_should_return_deleted_suppression(
+        self, suppressions_api: SuppressionsApi, sample_suppression_dict: dict
+    ) -> None:
+        responses.delete(
+            f"{BASE_SUPPRESSIONS_URL}/{SUPPRESSION_ID}",
+            json=sample_suppression_dict,
+            status=200,
+        )
+
+        deleted_suppression = suppressions_api.delete(SUPPRESSION_ID)
+
+        assert isinstance(deleted_suppression, Suppression)
+        assert deleted_suppression.id == SUPPRESSION_ID
+        assert deleted_suppression.type == SuppressionType.UNSUBSCRIPTION
+        assert deleted_suppression.email == "[email protected]"
+        assert deleted_suppression.sending_stream == SendingStream.TRANSACTIONAL