Fix issue #45: Add ContactEventsApi, related models, tests and examples

Author: Ihor Bilous

examples/contacts/contact_events.py

@@ -0,0 +1,34 @@
+import mailtrap as mt
+from mailtrap.models.contacts import ContactEvent
+
+API_TOKEN = "YOUR_API_TOKEN"
+ACCOUNT_ID = "YOUR_ACCOUNT_ID"
+
+client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID)
+contact_events_api = client.contacts_api.contact_events
+
+
+def create_contact_event(
+    contact_identifier: str,
+    contact_event_params: mt.ContactEventParams,
+) -> ContactEvent:
+    return contact_events_api.create(
+        contact_identifier=contact_identifier,
+        contact_event_params=contact_event_params,
+    )
+
+
+if __name__ == "__main__":
+    contact_event = create_contact_event(
+        contact_identifier="01988623-832f-79df-8aae-a480f8ff7249",
+        contact_event_params=mt.ContactEventParams(
+            name="UserLogin",
+            params={
+                "user_id": 101,
+                "user_name": "John Smith",
+                "is_active": True,
+                "last_seen": None,
+            },
+        ),
+    )
+    print(contact_event)

mailtrap/__init__.py

@@ -6,6 +6,7 @@
 from .exceptions import ClientConfigurationError
 from .exceptions import MailtrapError
 from .models.accounts import AccountAccessFilterParams
+from .models.contacts import ContactEventParams
 from .models.contacts import ContactExportFilter
 from .models.contacts import ContactListParams
 from .models.contacts import CreateContactExportParams

mailtrap/api/contacts.py

@@ -1,3 +1,4 @@
+from mailtrap.api.resources.contact_events import ContactEventsApi
 from mailtrap.api.resources.contact_exports import ContactExportsApi
 from mailtrap.api.resources.contact_fields import ContactFieldsApi
 from mailtrap.api.resources.contact_imports import ContactImportsApi
@@ -11,6 +12,10 @@ def __init__(self, client: HttpClient, account_id: str) -> None:
         self._account_id = account_id
         self._client = client
 
+    @property
+    def contact_events(self) -> ContactEventsApi:
+        return ContactEventsApi(account_id=self._account_id, client=self._client)
+
     @property
     def contact_exports(self) -> ContactExportsApi:
         return ContactExportsApi(account_id=self._account_id, client=self._client)

mailtrap/api/resources/contact_events.py

@@ -0,0 +1,26 @@
+from typing import Optional
+
+from mailtrap.http import HttpClient
+from mailtrap.models.contacts import ContactEvent
+from mailtrap.models.contacts import ContactEventParams
+
+
+class ContactEventsApi:
+    def __init__(self, client: HttpClient, account_id: str) -> None:
+        self._account_id = account_id
+        self._client = client
+
+    def create(
+        self,
+        contact_identifier: str,
+        contact_event_params: ContactEventParams,
+    ) -> ContactEvent:
+        """Create a new Contact Event"""
+        response = self._client.post(
+            self._api_path(contact_identifier),
+            json=contact_event_params.api_data,
+        )
+        return ContactEvent(**response)
+
+    def _api_path(self, contact_identifier: Optional[str] = None) -> str:
+        return f"/api/accounts/{self._account_id}/contacts/{contact_identifier}/events"

mailtrap/models/contacts.py

@@ -1,5 +1,6 @@
 from datetime import datetime
 from enum import Enum
+from typing import Any
 from typing import Optional
 from typing import Union
 
@@ -143,3 +144,17 @@ class ContactExportDetail:
     created_at: datetime
     updated_at: datetime
     url: Optional[str] = None
+
+
+@dataclass
+class ContactEventParams(RequestParams):
+    name: str
+    params: Optional[dict[str, Any]] = None
+
+
+@dataclass
+class ContactEvent:
+    contact_id: str
+    contact_email: str
+    name: str
+    params: dict[str, Any]

tests/unit/api/contacts/test_contact_events.py

@@ -0,0 +1,150 @@
+from typing import Any
+
+import pytest
+import responses
+
+from mailtrap.api.resources.contact_events import ContactEventsApi
+from mailtrap.config import GENERAL_HOST
+from mailtrap.exceptions import APIError
+from mailtrap.http import HttpClient
+from mailtrap.models.contacts import ContactEvent
+from mailtrap.models.contacts import ContactEventParams
+from tests import conftest
+
+ACCOUNT_ID = "321"
+EXPORT_ID = 1
+CONTACT_IDENTIFIER = "d82d0d9e-bbb7-4656-a591-e64682bffae7"
+BASE_CONTACT_EVENTS_URL = (
+    f"https://{GENERAL_HOST}/api/accounts/{ACCOUNT_ID}"
+    f"/contacts/{CONTACT_IDENTIFIER}/events"
+)
+
+
+@pytest.fixture
+def client() -> ContactEventsApi:
+    return ContactEventsApi(account_id=ACCOUNT_ID, client=HttpClient(GENERAL_HOST))
+
+
+@pytest.fixture
+def sample_contact_event_dict() -> dict[str, Any]:
+    return {
+        "contact_id": CONTACT_IDENTIFIER,
+        "contact_email": "[email protected]",
+        "name": "UserLogin",
+        "params": {
+            "user_id": 101,
+            "user_name": "John Smith",
+            "is_active": True,
+            "last_seen": None,
+        },
+    }
+
+
+@pytest.fixture
+def sample_create_contact_event_params() -> ContactEventParams:
+    return ContactEventParams(
+        name="UserLogin",
+        params={
+            "user_id": 101,
+            "user_name": "John Smith",
+            "is_active": True,
+            "last_seen": None,
+        },
+    )
+
+
+class TestContactEventsApi:
+
+    @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,
+            ),
+            (
+                conftest.RATE_LIMIT_ERROR_STATUS_CODE,
+                conftest.RATE_LIMIT_ERROR_RESPONSE,
+                conftest.RATE_LIMIT_ERROR_MESSAGE,
+            ),
+            (
+                conftest.INTERNAL_SERVER_ERROR_STATUS_CODE,
+                conftest.INTERNAL_SERVER_ERROR_RESPONSE,
+                conftest.INTERNAL_SERVER_ERROR_MESSAGE,
+            ),
+            (
+                422,
+                {
+                    "errors": {
+                        "name": [["must be a string", "is too long"]],
+                        "params": [
+                            [
+                                "must be a hash",
+                                "key 'foo' is too long",
+                                "value for 'bar' is too long",
+                            ]
+                        ],
+                    }
+                },
+                (
+                    "name: ['must be a string', 'is too long']; "
+                    "params: ['must be a hash', \"key 'foo' is too long\", "
+                    "\"value for 'bar' is too long\"]"
+                ),
+            ),
+        ],
+    )
+    @responses.activate
+    def test_create_should_raise_api_errors(
+        self,
+        client: ContactEventsApi,
+        status_code: int,
+        response_json: dict,
+        expected_error_message: str,
+        sample_create_contact_event_params: ContactEventParams,
+    ) -> None:
+        responses.post(
+            BASE_CONTACT_EVENTS_URL,
+            status=status_code,
+            json=response_json,
+        )
+
+        with pytest.raises(APIError) as exc_info:
+            client.create(CONTACT_IDENTIFIER, sample_create_contact_event_params)
+
+        print(str(exc_info.value))
+
+        assert expected_error_message in str(exc_info.value)
+
+    @responses.activate
+    def test_create_should_return_contact_event(
+        self,
+        client: ContactEventsApi,
+        sample_contact_event_dict: dict,
+        sample_create_contact_event_params: ContactEventParams,
+    ) -> None:
+        responses.post(
+            BASE_CONTACT_EVENTS_URL,
+            json=sample_contact_event_dict,
+            status=201,
+        )
+
+        result = client.create(CONTACT_IDENTIFIER, sample_create_contact_event_params)
+
+        assert isinstance(result, ContactEvent)
+        assert result.contact_id == CONTACT_IDENTIFIER
+        assert result.contact_email == "[email protected]"
+        assert result.name == "UserLogin"
+        request = responses.calls[0].request
+        assert request.url == BASE_CONTACT_EVENTS_URL

tests/unit/models/test_contacts.py

@@ -1,5 +1,6 @@
 import pytest
 
+from mailtrap.models.contacts import ContactEventParams
 from mailtrap.models.contacts import ContactExportFilter
 from mailtrap.models.contacts import ContactListParams
 from mailtrap.models.contacts import CreateContactExportParams
@@ -198,3 +199,35 @@ def test_create_contact_export_params_with_empty_filters_should_work(self) -> No
         api_data = params.api_data
 
         assert api_data == {"filters": []}
+
+
+class TestContactEventParams:
+    def test_contact_event_params_api_data_should_return_correct_dict(
+        self,
+    ) -> None:
+        params = ContactEventParams(
+            name="UserLogin",
+            params={
+                "user_id": 101,
+                "user_name": "John Smith",
+                "is_active": True,
+                "last_seen": None,
+            },
+        )
+
+        api_data = params.api_data
+        assert api_data == {
+            "name": "UserLogin",
+            "params": {
+                "user_id": 101,
+                "user_name": "John Smith",
+                "is_active": True,
+                "last_seen": None,
+            },
+        }
+
+    def test_contact_event_params_with_empty_params_should_work(self) -> None:
+        """Test that empty filters list works correctly."""
+        params = ContactEventParams(name="UserLogin")
+        api_data = params.api_data
+        assert api_data == {"name": "UserLogin"}