From 356e4b6fb53e4701f5ec343f6ae615f8f79551c4 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Sun, 15 Mar 2026 19:49:25 +0000 Subject: [PATCH 1/2] Fix API docs and repository links --- CHANGELOG.md | 29 +++++++++++++++++------------ README.md | 2 +- mailtrap/client.py | 2 +- pyproject.toml | 6 +++--- 4 files changed, 22 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 609b6a1..a21c72b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,30 +1,34 @@ ## [2.4.0] - 2025-12-04 + * Fix issue #52: Update README.md using new guideline by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/55 * Fix issue #53: Add full usage in all examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/56 * Merge functionality and examples in Readme by @yanchuk in https://github.com/mailtrap/mailtrap-python/pull/57 * Fix issue #54: Add SendingDomainsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/58 ## [2.3.0] - 2025-10-24 + * Fix issue #24: Add batch_send method to SendingApi, add models by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/47 * Fix issue #42: Add GeneralApi, related models, examples, tests. by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/48 * Fix issue #41: Add ContactExportsApi, related models, tests and examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/49 * Fix issue #45: Add ContactEventsApi, related models, tests and examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/51 ## [2.2.0] - 2025-09-18 -- Potential fix for code scanning alert no. 1: Workflow does not contain permissions by @mklocek in https://github.com/railsware/mailtrap-python/pull/15 -- Fix issue #29. Add support of Emails Sandbox (Testing) API: Projects by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/31 -- Issue 25 by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/33 -- Fix issue #18: Add api for EmailTemplates, add tests and examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/34 -- Fix issue #19: Add ContactFieldsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/35 -- Fix issue #20: Add ContactListsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/36 -- Fix issue #21: Add ContactsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/37 -- Fix issue #22: Add ContactImportsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/38 -- Fix issue #23: Add SuppressionsApi, related models, tests and examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/39 -- Fix issue #27: Add InboxesApi, related models, tests, examples. by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/40 -- Fix issue #26: Add MessagesApi, releated models, examples, tests by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/43 -- Fix issue #28: Add AttachmentsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/railsware/mailtrap-python/pull/44 + +- Potential fix for code scanning alert no. 1: Workflow does not contain permissions by @mklocek in https://github.com/mailtrap/mailtrap-python/pull/15 +- Fix issue #29. Add support of Emails Sandbox (Testing) API: Projects by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/31 +- Issue 25 by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/33 +- Fix issue #18: Add api for EmailTemplates, add tests and examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/34 +- Fix issue #19: Add ContactFieldsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/35 +- Fix issue #20: Add ContactListsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/36 +- Fix issue #21: Add ContactsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/37 +- Fix issue #22: Add ContactImportsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/38 +- Fix issue #23: Add SuppressionsApi, related models, tests and examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/39 +- Fix issue #27: Add InboxesApi, related models, tests, examples. by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/40 +- Fix issue #26: Add MessagesApi, releated models, examples, tests by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/43 +- Fix issue #28: Add AttachmentsApi, related models, tests, examples by @Ihor-Bilous in https://github.com/mailtrap/mailtrap-python/pull/44 ## [2.1.0] - 2025-05-12 + - Add sandbox mode support in MailtrapClient - It requires inbox_id parameter to be set - Add bulk mode support in MailtrapClient @@ -32,6 +36,7 @@ - Add support for python 3.12 - 3.13 ## [2.0.1] - 2023-05-18 + - Add User-Agent header to all requests ## [2.0.0] - 2023-03-11 diff --git a/README.md b/README.md index 9dc76d2..cb37127 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ # Official Mailtrap Python client -This Python package offers integration with the [official API](https://api-docs.mailtrap.io/) for [Mailtrap](https://mailtrap.io). +This Python package offers integration with the [official API](https://docs.mailtrap.io/developers) for [Mailtrap](https://mailtrap.io). Add email sending functionality to your Python application quickly with Mailtrap. diff --git a/mailtrap/client.py b/mailtrap/client.py index 17378e5..142f2d9 100644 --- a/mailtrap/client.py +++ b/mailtrap/client.py @@ -37,7 +37,7 @@ class MailtrapClient: SANDBOX_HOST = SANDBOX_HOST DEFAULT_USER_AGENT = ( f"mailtrap-python/{importlib.metadata.version('mailtrap')} " - "(https://github.com/railsware/mailtrap-python)" + "(https://github.com/mailtrap/mailtrap-python)" ) def __init__( diff --git a/pyproject.toml b/pyproject.toml index 938e0a0..2aa398e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,9 +21,9 @@ dynamic = ["dependencies"] [project.urls] Homepage = "https://mailtrap.io/" -Documentation = "https://github.com/railsware/mailtrap-python" -Repository = "https://github.com/railsware/mailtrap-python.git" -"API documentation" = "https://api-docs.mailtrap.io/" +Documentation = "https://github.com/mailtrap/mailtrap-python" +Repository = "https://github.com/mailtrap/mailtrap-python.git" +"API documentation" = "https://docs.mailtrap.io/developers" [build-system] requires = ["setuptools"] From 7912137c106f243e3353ca82a33ad97ac5d4ce48 Mon Sep 17 00:00:00 2001 From: Marcin Klocek Date: Sun, 15 Mar 2026 20:20:35 +0000 Subject: [PATCH 2/2] Add support for Email Logs API --- README.md | 3 + examples/email_logs/email_logs.py | 61 +++ mailtrap/__init__.py | 3 + mailtrap/api/email_logs.py | 12 + mailtrap/api/resources/email_logs.py | 48 +++ mailtrap/client.py | 21 +- mailtrap/models/email_logs.py | 383 +++++++++++++++++++ tests/unit/api/email_logs/test_email_logs.py | 202 ++++++++++ tests/unit/models/test_email_log_models.py | 214 +++++++++++ 9 files changed, 941 insertions(+), 6 deletions(-) create mode 100644 examples/email_logs/email_logs.py create mode 100644 mailtrap/api/email_logs.py create mode 100644 mailtrap/api/resources/email_logs.py create mode 100644 mailtrap/models/email_logs.py create mode 100644 tests/unit/api/email_logs/test_email_logs.py create mode 100644 tests/unit/models/test_email_log_models.py diff --git a/README.md b/README.md index cb37127..bade01c 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,9 @@ The same situation applies to both `client.batch_send()` and `client.sending_api ### Suppressions API: - Suppressions (find & delete) – [`suppressions/suppressions.py`](examples/suppressions/suppressions.py) +### Email Logs API: +- List email logs (with filters & pagination) and get message by ID – [`email_logs/email_logs.py`](examples/email_logs/email_logs.py) + ### General API: - Account Accesses management – [`general/account_accesses.py`](examples/general/account_accesses.py) - Accounts info – [`general/accounts.py`](examples/general/accounts.py) diff --git a/examples/email_logs/email_logs.py b/examples/email_logs/email_logs.py new file mode 100644 index 0000000..f8933e5 --- /dev/null +++ b/examples/email_logs/email_logs.py @@ -0,0 +1,61 @@ +"""Example: List email logs and get a single message by ID.""" + +from datetime import datetime +from datetime import timedelta +from datetime import timezone + +import mailtrap as mt +from mailtrap.models.email_logs import EmailLogsListFilters +from mailtrap.models.email_logs import filter_ci_equal +from mailtrap.models.email_logs import filter_string_equal +from mailtrap.models.email_logs import filter_string_not_empty + +API_TOKEN = "YOUR_API_TOKEN" +ACCOUNT_ID = "YOUR_ACCOUNT_ID" + +client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID) +email_logs_api = client.email_logs_api.email_logs + + +def list_email_logs(): + """List email logs (first page).""" + return email_logs_api.get_list() + + +def list_email_logs_with_filters(): + """List email logs from last 2 days, by category(s), with non-empty subject.""" + now = datetime.now(timezone.utc) + two_days_ago = now - timedelta(days=2) + filters = EmailLogsListFilters( + sent_after=two_days_ago.isoformat().replace("+00:00", "Z"), + sent_before=now.isoformat().replace("+00:00", "Z"), + subject=filter_string_not_empty(), + to=filter_ci_equal("recipient@example.com"), + category=filter_string_equal(["Welcome Email", "Password Reset"]), + ) + return email_logs_api.get_list(filters=filters) + + +def get_next_page(previous_response): + """Fetch next page using cursor from previous response.""" + if previous_response.next_page_cursor is None: + return None + return email_logs_api.get_list(search_after=previous_response.next_page_cursor) + + +def get_message(message_id: str): + """Get a single email log message by UUID.""" + return email_logs_api.get_by_id(message_id) + + +if __name__ == "__main__": + # List first page + response = list_email_logs() + print(f"Total: {response.total_count}, messages: {len(response.messages)}") + for msg in response.messages: + print(f" {msg.message_id} | {msg.from_} -> {msg.to} | {msg.status}") + + # Get single message + if response.messages: + detail = get_message(response.messages[0].message_id) + print(f"Detail: {detail.subject}, events: {len(detail.events)}") diff --git a/mailtrap/__init__.py b/mailtrap/__init__.py index 67b5832..e6805d4 100644 --- a/mailtrap/__init__.py +++ b/mailtrap/__init__.py @@ -15,6 +15,9 @@ from .models.contacts import ImportContactParams from .models.contacts import UpdateContactFieldParams from .models.contacts import UpdateContactParams +from .models.email_logs import EmailLogMessage +from .models.email_logs import EmailLogsListFilters +from .models.email_logs import EmailLogsListResponse from .models.inboxes import CreateInboxParams from .models.inboxes import UpdateInboxParams from .models.mail import Address diff --git a/mailtrap/api/email_logs.py b/mailtrap/api/email_logs.py new file mode 100644 index 0000000..af2aee5 --- /dev/null +++ b/mailtrap/api/email_logs.py @@ -0,0 +1,12 @@ +from mailtrap.api.resources.email_logs import EmailLogsApi +from mailtrap.http import HttpClient + + +class EmailLogsBaseApi: + def __init__(self, client: HttpClient, account_id: str) -> None: + self._account_id = account_id + self._client = client + + @property + def email_logs(self) -> EmailLogsApi: + return EmailLogsApi(client=self._client, account_id=self._account_id) diff --git a/mailtrap/api/resources/email_logs.py b/mailtrap/api/resources/email_logs.py new file mode 100644 index 0000000..1d23cc5 --- /dev/null +++ b/mailtrap/api/resources/email_logs.py @@ -0,0 +1,48 @@ +"""Email Logs API resource – list and get email sending logs.""" + +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.email_logs import EmailLogMessage +from mailtrap.models.email_logs import EmailLogsListFilters +from mailtrap.models.email_logs import EmailLogsListResponse + + +class EmailLogsApi: + def __init__(self, client: HttpClient, account_id: str) -> None: + self._account_id = account_id + self._client = client + + def get_list( + self, + filters: Optional[EmailLogsListFilters] = None, + search_after: Optional[str] = None, + ) -> EmailLogsListResponse: + """ + List email logs (paginated). Results are ordered by sent_at descending. + Use search_after with next_page_cursor from the previous response for + the next page. + """ + params: dict[str, object] = {} + if filters is not None: + params.update(filters.to_params()) + if search_after is not None: + params["search_after"] = search_after + response = self._client.get(self._api_path(), params=params or None) + messages = [EmailLogMessage.from_api(msg) for msg in response.get("messages", [])] + return EmailLogsListResponse( + messages=messages, + total_count=response.get("total_count", 0), + next_page_cursor=response.get("next_page_cursor"), + ) + + def get_by_id(self, sending_message_id: str) -> EmailLogMessage: + """Get a single email log message by its UUID.""" + response = self._client.get(self._api_path(sending_message_id)) + return EmailLogMessage.from_api(response) + + def _api_path(self, sending_message_id: Optional[str] = None) -> str: + path = f"/api/accounts/{self._account_id}/email_logs" + if sending_message_id is not None: + path = f"{path}/{sending_message_id}" + return path diff --git a/mailtrap/client.py b/mailtrap/client.py index 142f2d9..866f6f0 100644 --- a/mailtrap/client.py +++ b/mailtrap/client.py @@ -7,6 +7,7 @@ from pydantic import TypeAdapter from mailtrap.api.contacts import ContactsBaseApi +from mailtrap.api.email_logs import EmailLogsBaseApi from mailtrap.api.general import GeneralApi from mailtrap.api.sending import SendingApi from mailtrap.api.sending_domains import SendingDomainsBaseApi @@ -81,7 +82,7 @@ def testing_api(self) -> TestingApi: @property def email_templates_api(self) -> EmailTemplatesApi: - self._validate_account_id() + self._validate_account_id("Email Templates API") return EmailTemplatesApi( account_id=cast(str, self.account_id), client=HttpClient(host=GENERAL_HOST, headers=self.headers), @@ -89,7 +90,7 @@ def email_templates_api(self) -> EmailTemplatesApi: @property def contacts_api(self) -> ContactsBaseApi: - self._validate_account_id() + self._validate_account_id("Contacts API") return ContactsBaseApi( account_id=cast(str, self.account_id), client=HttpClient(host=GENERAL_HOST, headers=self.headers), @@ -97,7 +98,7 @@ def contacts_api(self) -> ContactsBaseApi: @property def suppressions_api(self) -> SuppressionsBaseApi: - self._validate_account_id() + self._validate_account_id("Suppressions API") return SuppressionsBaseApi( account_id=cast(str, self.account_id), client=HttpClient(host=GENERAL_HOST, headers=self.headers), @@ -105,12 +106,20 @@ def suppressions_api(self) -> SuppressionsBaseApi: @property def sending_domains_api(self) -> SendingDomainsBaseApi: - self._validate_account_id() + self._validate_account_id("Sending Domains API") return SendingDomainsBaseApi( account_id=cast(str, self.account_id), client=HttpClient(host=GENERAL_HOST, headers=self.headers), ) + @property + def email_logs_api(self) -> EmailLogsBaseApi: + self._validate_account_id("Email Logs API") + return EmailLogsBaseApi( + 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) @@ -169,9 +178,9 @@ def _sending_api_host(self) -> str: return BULK_HOST return SENDING_HOST - def _validate_account_id(self) -> None: + def _validate_account_id(self, api_name: str = "Testing API") -> None: if not self.account_id: - raise ClientConfigurationError("`account_id` is required for Testing API") + raise ClientConfigurationError(f"`account_id` is required for {api_name}") def _validate_itself(self) -> None: if self.sandbox and not self.inbox_id: diff --git a/mailtrap/models/email_logs.py b/mailtrap/models/email_logs.py new file mode 100644 index 0000000..dc1cb00 --- /dev/null +++ b/mailtrap/models/email_logs.py @@ -0,0 +1,383 @@ +"""Models for Email Logs API (list, get message, filters).""" + +from typing import Any +from typing import Literal +from typing import Optional +from typing import Union + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic.dataclasses import dataclass + +# --- Event details (for get-by-id) --- + + +@dataclass +class EventDetailsDelivery: + sending_ip: Optional[str] = None + recipient_mx: Optional[str] = None + email_service_provider: Optional[str] = None + + +@dataclass +class EventDetailsOpen: + web_ip_address: Optional[str] = None + + +@dataclass +class EventDetailsClick: + click_url: Optional[str] = None + web_ip_address: Optional[str] = None + + +@dataclass +class EventDetailsBounce: + sending_ip: Optional[str] = None + recipient_mx: Optional[str] = None + email_service_provider: Optional[str] = None + email_service_provider_status: Optional[str] = None + email_service_provider_response: Optional[str] = None + bounce_category: Optional[str] = None + + +@dataclass +class EventDetailsSpam: + spam_feedback_type: Optional[str] = None + + +@dataclass +class EventDetailsUnsubscribe: + web_ip_address: Optional[str] = None + + +@dataclass +class EventDetailsReject: + reject_reason: Optional[str] = None + + +@dataclass +class MessageEvent: + """Event attached to a message (delivery, open, click, bounce, etc.).""" + + event_type: str + created_at: str + details: Optional[ + Union[ + EventDetailsDelivery, + EventDetailsOpen, + EventDetailsClick, + EventDetailsBounce, + EventDetailsSpam, + EventDetailsUnsubscribe, + EventDetailsReject, + dict[str, Any], + ] + ] = None + + +def _parse_event_details( + event_type: str, + data: Optional[dict[str, Any]], +) -> Optional[ + Union[ + EventDetailsDelivery, + EventDetailsOpen, + EventDetailsClick, + EventDetailsBounce, + EventDetailsSpam, + EventDetailsUnsubscribe, + EventDetailsReject, + dict[str, Any], + ] +]: + """Build the correct EventDetails* from event_type and raw details dict.""" + if data is None: + return None + if event_type == "delivery": + return EventDetailsDelivery(**data) + if event_type == "open": + return EventDetailsOpen(**data) + if event_type == "click": + return EventDetailsClick(**data) + if event_type in ("soft_bounce", "bounce"): + return EventDetailsBounce(**data) + if event_type == "spam": + return EventDetailsSpam(**data) + if event_type == "unsubscribe": + return EventDetailsUnsubscribe(**data) + if event_type in ("suspension", "reject"): + return EventDetailsReject(**data) + return data + + +def _parse_message_event(event_dict: dict[str, Any]) -> MessageEvent: + """Build MessageEvent from API dict using event_type to select details type.""" + event_type = event_dict["event_type"] + created_at = event_dict["created_at"] + details = _parse_event_details(event_type, event_dict.get("details")) + return MessageEvent( + event_type=event_type, + created_at=created_at, + details=details, + ) + + +class EmailLogMessage(BaseModel): + """ + Email log message. Used for both list and get-by-id; from list response + only summary fields are set, raw_message_url and events are empty. + """ + + model_config = ConfigDict(populate_by_name=True) + + message_id: str + status: Literal["delivered", "not_delivered", "enqueued", "opted_out"] + subject: Optional[str] = None + from_: str = Field(alias="from") + to: str + sent_at: str + client_ip: Optional[str] = None + category: Optional[str] = None + custom_variables: dict[str, Any] = Field(default_factory=dict) + sending_stream: Literal["transactional", "bulk"] + sending_domain_id: int + template_id: Optional[int] = None + template_variables: dict[str, Any] = Field(default_factory=dict) + opens_count: int + clicks_count: int + raw_message_url: Optional[str] = None + events: list[MessageEvent] = Field(default_factory=list) + + @classmethod + def from_api(cls, data: dict[str, Any]) -> "EmailLogMessage": + """Build from API response (handles 'from' alias and None for events/vars).""" + payload = dict(data) + if payload.get("events") is None: + payload["events"] = [] + else: + payload["events"] = [_parse_message_event(e) for e in payload["events"]] + if payload.get("custom_variables") is None: + payload["custom_variables"] = {} + if payload.get("template_variables") is None: + payload["template_variables"] = {} + return cls(**payload) + + +@dataclass +class EmailLogsListResponse: + """Paginated response from list email logs.""" + + messages: list[EmailLogMessage] + total_count: int + next_page_cursor: Optional[str] + + +# --- Filters for list (operator + value) --- + +# Status and events +StatusValue = Literal["delivered", "not_delivered", "enqueued", "opted_out"] +EventsFilterValue = Literal[ + "delivery", + "open", + "click", + "bounce", + "spam", + "unsubscribe", + "soft_bounce", + "reject", + "suspension", +] + +# Numeric comparison +FilterOperatorNumeric = Literal["equal", "greater_than", "less_than"] + +# Sending stream +SendingStreamValue = Literal["transactional", "bulk"] + + +def _filter_spec( + operator: str, + value: Optional[Union[str, int, list[Any]]] = None, +) -> dict[str, Any]: + """Build a filter spec dict for API (operator + optional value).""" + out: dict[str, Any] = {"operator": operator} + if value is not None: + out["value"] = value + return out + + +# Convenience filter builders (optional; users can also pass raw dicts) + + +def filter_ci_equal(value: Union[str, list[str]]) -> dict[str, Any]: + return _filter_spec("ci_equal", value) + + +def filter_ci_not_equal(value: Union[str, list[str]]) -> dict[str, Any]: + return _filter_spec("ci_not_equal", value) + + +def filter_ci_contain(value: str) -> dict[str, Any]: + return _filter_spec("ci_contain", value) + + +def filter_ci_not_contain(value: str) -> dict[str, Any]: + return _filter_spec("ci_not_contain", value) + + +def filter_status_equal(value: Union[StatusValue, list[StatusValue]]) -> dict[str, Any]: + return _filter_spec("equal", value) + + +def filter_status_not_equal( + value: Union[StatusValue, list[StatusValue]], +) -> dict[str, Any]: + return _filter_spec("not_equal", value) + + +def filter_events_include( + value: Union[EventsFilterValue, list[EventsFilterValue]], +) -> dict[str, Any]: + return _filter_spec("include_event", value) + + +def filter_events_not_include( + value: Union[EventsFilterValue, list[EventsFilterValue]], +) -> dict[str, Any]: + return _filter_spec("not_include_event", value) + + +def filter_numeric( + operator: FilterOperatorNumeric, + value: int, +) -> dict[str, Any]: + return _filter_spec(operator, value) + + +def filter_sending_domain_id_equal(value: Union[int, list[int]]) -> dict[str, Any]: + return _filter_spec("equal", value) + + +def filter_sending_domain_id_not_equal(value: Union[int, list[int]]) -> dict[str, Any]: + return _filter_spec("not_equal", value) + + +def filter_sending_stream_equal( + value: Union[SendingStreamValue, list[SendingStreamValue]], +) -> dict[str, Any]: + return _filter_spec("equal", value) + + +def filter_sending_stream_not_equal( + value: Union[SendingStreamValue, list[SendingStreamValue]], +) -> dict[str, Any]: + return _filter_spec("not_equal", value) + + +def filter_string_equal(value: Union[str, list[str]]) -> dict[str, Any]: + return _filter_spec("equal", value) + + +def filter_string_not_equal(value: Union[str, list[str]]) -> dict[str, Any]: + return _filter_spec("not_equal", value) + + +def filter_empty() -> dict[str, Any]: + return _filter_spec("empty") + + +def filter_string_not_empty() -> dict[str, Any]: + return _filter_spec("not_empty") + + +# --- Email logs list filters (top-level) --- + + +class EmailLogsListFilters: + """ + Filters for listing email logs. Pass to email_logs.get_list(filters=...). + All fields are optional. Date range: sent_after, sent_before (ISO 8601). + Other fields are filter specs: {"operator": "...", "value": ...} or + {"operator": "empty"}. + """ + + def __init__( + self, + *, + sent_after: Optional[str] = None, + sent_before: Optional[str] = None, + to: Optional[dict[str, Any]] = None, + from_: Optional[dict[str, Any]] = None, + subject: Optional[dict[str, Any]] = None, + status: Optional[dict[str, Any]] = None, + events: Optional[dict[str, Any]] = None, + clicks_count: Optional[dict[str, Any]] = None, + opens_count: Optional[dict[str, Any]] = None, + client_ip: Optional[dict[str, Any]] = None, + sending_ip: Optional[dict[str, Any]] = None, + email_service_provider_response: Optional[dict[str, Any]] = None, + email_service_provider: Optional[dict[str, Any]] = None, + recipient_mx: Optional[dict[str, Any]] = None, + category: Optional[dict[str, Any]] = None, + sending_domain_id: Optional[dict[str, Any]] = None, + sending_stream: Optional[dict[str, Any]] = None, + ) -> None: + self.sent_after = sent_after + self.sent_before = sent_before + self.to = to + self.from_ = from_ + self.subject = subject + self.status = status + self.events = events + self.clicks_count = clicks_count + self.opens_count = opens_count + self.client_ip = client_ip + self.sending_ip = sending_ip + self.email_service_provider_response = email_service_provider_response + self.email_service_provider = email_service_provider + self.recipient_mx = recipient_mx + self.category = category + self.sending_domain_id = sending_domain_id + self.sending_stream = sending_stream + + def to_params(self) -> dict[str, Any]: + """Serialize to query params: filters[key] and filters[key][operator].""" + params: dict[str, Any] = {} + if self.sent_after is not None: + params["filters[sent_after]"] = self.sent_after + if self.sent_before is not None: + params["filters[sent_before]"] = self.sent_before + + # Map Python-friendly name to API key (from_ -> "from") + spec_keys = [ + ("to", self.to), + ("from", self.from_), + ("subject", self.subject), + ("status", self.status), + ("events", self.events), + ("clicks_count", self.clicks_count), + ("opens_count", self.opens_count), + ("client_ip", self.client_ip), + ("sending_ip", self.sending_ip), + ("email_service_provider_response", self.email_service_provider_response), + ("email_service_provider", self.email_service_provider), + ("recipient_mx", self.recipient_mx), + ("category", self.category), + ("sending_domain_id", self.sending_domain_id), + ("sending_stream", self.sending_stream), + ] + for key, spec in spec_keys: + if spec is None: + continue + prefix = f"filters[{key}]" + if "operator" in spec: + params[f"{prefix}[operator]"] = spec["operator"] + if "value" in spec: + val = spec["value"] + if isinstance(val, list): + # requests serializes list as key=v1&key=v2 + params[f"{prefix}[value][]"] = val + else: + params[f"{prefix}[value]"] = val + return params diff --git a/tests/unit/api/email_logs/test_email_logs.py b/tests/unit/api/email_logs/test_email_logs.py new file mode 100644 index 0000000..87531db --- /dev/null +++ b/tests/unit/api/email_logs/test_email_logs.py @@ -0,0 +1,202 @@ +"""Unit tests for Email Logs API.""" + +from typing import Any + +import pytest +import responses + +from mailtrap.api.resources.email_logs import EmailLogsApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.email_logs import EmailLogMessage +from mailtrap.models.email_logs import EmailLogsListFilters +from tests import conftest + +ACCOUNT_ID = "321" +MESSAGE_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890" +BASE_EMAIL_LOGS_URL = f"https://{GENERAL_HOST}/api/accounts/{ACCOUNT_ID}/email_logs" + + +@pytest.fixture +def email_logs_api() -> EmailLogsApi: + return EmailLogsApi(client=HttpClient(GENERAL_HOST), account_id=ACCOUNT_ID) + + +@pytest.fixture +def sample_message_dict() -> dict[str, Any]: + return { + "message_id": MESSAGE_ID, + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": "203.0.113.42", + "category": "Welcome Email", + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": 100, + "template_variables": {}, + "opens_count": 2, + "clicks_count": 1, + } + + +@pytest.fixture +def sample_list_response(sample_message_dict: dict[str, Any]) -> dict[str, Any]: + return { + "messages": [sample_message_dict], + "total_count": 1, + "next_page_cursor": None, + } + + +@pytest.fixture +def sample_message_detail(sample_message_dict: dict[str, Any]) -> dict[str, Any]: + detail = dict(sample_message_dict) + detail["raw_message_url"] = "https://storage.example.com/signed/eml/..." + detail["events"] = [ + { + "event_type": "click", + "created_at": "2025-01-15T10:35:00Z", + "details": { + "click_url": "https://example.com/track/abc", + "web_ip_address": "198.51.100.50", + }, + }, + ] + return detail + + +class TestEmailLogsApi: + @responses.activate + def test_get_list_returns_response( + self, + email_logs_api: EmailLogsApi, + sample_list_response: dict[str, Any], + ) -> None: + responses.get(BASE_EMAIL_LOGS_URL, json=sample_list_response, status=200) + + result = email_logs_api.get_list() + + assert result.total_count == 1 + assert result.next_page_cursor is None + assert len(result.messages) == 1 + msg = result.messages[0] + assert isinstance(msg, EmailLogMessage) + assert msg.message_id == MESSAGE_ID + assert msg.status == "delivered" + assert msg.from_ == "sender@example.com" + assert msg.to == "recipient@example.com" + assert msg.opens_count == 2 + assert msg.clicks_count == 1 + + @responses.activate + def test_get_list_with_filters_and_search_after( + self, + email_logs_api: EmailLogsApi, + sample_list_response: dict[str, Any], + ) -> None: + filters = EmailLogsListFilters( + sent_after="2025-01-01T00:00:00Z", + sent_before="2025-01-31T23:59:59Z", + to={"operator": "ci_equal", "value": "recipient@example.com"}, + ) + responses.get( + BASE_EMAIL_LOGS_URL, + json=sample_list_response, + status=200, + ) + + result = email_logs_api.get_list(filters=filters, search_after="b2c3") + + assert result.total_count == 1 + assert len(result.messages) == 1 + assert result.messages[0].message_id == MESSAGE_ID + + @responses.activate + def test_get_by_id_returns_sending_message( + self, + email_logs_api: EmailLogsApi, + sample_message_detail: dict[str, Any], + ) -> None: + responses.get( + f"{BASE_EMAIL_LOGS_URL}/{MESSAGE_ID}", + json=sample_message_detail, + status=200, + ) + + msg = email_logs_api.get_by_id(MESSAGE_ID) + + assert isinstance(msg, EmailLogMessage) + assert msg.message_id == MESSAGE_ID + assert msg.from_ == "sender@example.com" + assert msg.raw_message_url is not None + assert len(msg.events) == 1 + assert msg.events[0].event_type == "click" + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", + [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.NOT_FOUND_STATUS_CODE, + conftest.NOT_FOUND_RESPONSE, + conftest.NOT_FOUND_ERROR_MESSAGE, + ), + ], + ) + @responses.activate + def test_get_list_raises_api_errors( + self, + email_logs_api: EmailLogsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get(BASE_EMAIL_LOGS_URL, status=status_code, json=response_json) + + with pytest.raises(APIError) as exc_info: + email_logs_api.get_list() + + assert expected_error_message in str(exc_info.value) + + @pytest.mark.parametrize( + "status_code,response_json,expected_error_message", + [ + ( + conftest.UNAUTHORIZED_STATUS_CODE, + conftest.UNAUTHORIZED_RESPONSE, + conftest.UNAUTHORIZED_ERROR_MESSAGE, + ), + ( + conftest.NOT_FOUND_STATUS_CODE, + conftest.NOT_FOUND_RESPONSE, + conftest.NOT_FOUND_ERROR_MESSAGE, + ), + ], + ) + @responses.activate + def test_get_by_id_raises_api_errors( + self, + email_logs_api: EmailLogsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get( + f"{BASE_EMAIL_LOGS_URL}/{MESSAGE_ID}", + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + email_logs_api.get_by_id(MESSAGE_ID) + + assert expected_error_message in str(exc_info.value) diff --git a/tests/unit/models/test_email_log_models.py b/tests/unit/models/test_email_log_models.py new file mode 100644 index 0000000..6f4b9e5 --- /dev/null +++ b/tests/unit/models/test_email_log_models.py @@ -0,0 +1,214 @@ +"""Unit tests for email logs models.""" + +from typing import Any + +from mailtrap.models.email_logs import EmailLogMessage +from mailtrap.models.email_logs import EmailLogsListFilters +from mailtrap.models.email_logs import EventDetailsBounce +from mailtrap.models.email_logs import EventDetailsClick +from mailtrap.models.email_logs import EventDetailsDelivery +from mailtrap.models.email_logs import EventDetailsOpen +from mailtrap.models.email_logs import EventDetailsUnsubscribe +from mailtrap.models.email_logs import MessageEvent +from mailtrap.models.email_logs import filter_ci_equal +from mailtrap.models.email_logs import filter_sending_domain_id_equal + + +class TestEmailLogMessage: + def test_parses_list_item_with_from_key(self) -> None: + """List response has no raw_message_url or events; those stay empty.""" + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": "203.0.113.42", + "category": "Welcome Email", + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": 100, + "template_variables": {}, + "opens_count": 2, + "clicks_count": 1, + } + msg = EmailLogMessage(**data) + assert msg.message_id == data["message_id"] + assert msg.from_ == "sender@example.com" + assert msg.to == "recipient@example.com" + assert msg.status == "delivered" + assert msg.raw_message_url is None + assert msg.events == [] + + def test_from_api_handles_from_and_events(self) -> None: + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": None, + "category": None, + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": None, + "template_variables": {}, + "opens_count": 0, + "clicks_count": 0, + "raw_message_url": "https://example.com/raw", + "events": None, + } + msg = EmailLogMessage.from_api(data) + assert msg.from_ == "sender@example.com" + assert msg.raw_message_url == "https://example.com/raw" + assert msg.events == [] + + def test_from_api_parses_events_with_deterministic_details_type(self) -> None: + """event_type selects correct EventDetails* (open vs unsubscribe same shape).""" + # open and unsubscribe both have web_ip_address; selection is by event_type + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": None, + "category": None, + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": None, + "template_variables": {}, + "opens_count": 0, + "clicks_count": 0, + "raw_message_url": None, + "events": [ + { + "event_type": "open", + "created_at": "2025-01-15T10:35:00Z", + "details": {"web_ip_address": "198.51.100.50"}, + }, + { + "event_type": "unsubscribe", + "created_at": "2025-01-15T10:40:00Z", + "details": {"web_ip_address": "198.51.100.50"}, + }, + ], + } + msg = EmailLogMessage.from_api(data) + assert len(msg.events) == 2 + assert msg.events[0].event_type == "open" + assert isinstance(msg.events[0].details, EventDetailsOpen) + assert msg.events[0].details.web_ip_address == "198.51.100.50" + assert msg.events[1].event_type == "unsubscribe" + assert isinstance(msg.events[1].details, EventDetailsUnsubscribe) + assert msg.events[1].details.web_ip_address == "198.51.100.50" + + def test_from_api_parses_click_and_bounce_events(self) -> None: + """Click and bounce events get EventDetailsClick and EventDetailsBounce.""" + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": None, + "category": None, + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": None, + "template_variables": {}, + "opens_count": 0, + "clicks_count": 1, + "raw_message_url": None, + "events": [ + { + "event_type": "click", + "created_at": "2025-01-15T10:35:00Z", + "details": { + "click_url": "https://example.com/track/abc", + "web_ip_address": "198.51.100.50", + }, + }, + { + "event_type": "bounce", + "created_at": "2025-01-15T10:36:00Z", + "details": { + "email_service_provider_response": "User unknown", + "bounce_category": "invalid_recipient", + }, + }, + ], + } + msg = EmailLogMessage.from_api(data) + assert len(msg.events) == 2 + assert isinstance(msg.events[0].details, EventDetailsClick) + assert msg.events[0].details.click_url == "https://example.com/track/abc" + assert isinstance(msg.events[1].details, EventDetailsBounce) + assert msg.events[1].details.bounce_category == "invalid_recipient" + + def test_from_api_parses_delivery_event(self) -> None: + """Delivery event gets EventDetailsDelivery.""" + data: dict[str, Any] = { + "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "status": "delivered", + "subject": "Welcome", + "from": "sender@example.com", + "to": "recipient@example.com", + "sent_at": "2025-01-15T10:30:00Z", + "client_ip": None, + "category": None, + "custom_variables": {}, + "sending_stream": "transactional", + "sending_domain_id": 3938, + "template_id": None, + "template_variables": {}, + "opens_count": 0, + "clicks_count": 0, + "raw_message_url": None, + "events": [ + { + "event_type": "delivery", + "created_at": "2025-01-15T10:31:00Z", + "details": { + "email_service_provider": "Google", + "recipient_mx": "gmail-smtp-in.l.google.com", + }, + }, + ], + } + msg = EmailLogMessage.from_api(data) + assert len(msg.events) == 1 + assert isinstance(msg.events[0], MessageEvent) + assert msg.events[0].event_type == "delivery" + assert isinstance(msg.events[0].details, EventDetailsDelivery) + assert msg.events[0].details.email_service_provider == "Google" + + +class TestEmailLogsListFilters: + def test_to_params_date_range(self) -> None: + f = EmailLogsListFilters( + sent_after="2025-01-01T00:00:00Z", + sent_before="2025-01-31T23:59:59Z", + ) + params = f.to_params() + assert params["filters[sent_after]"] == "2025-01-01T00:00:00Z" + assert params["filters[sent_before]"] == "2025-01-31T23:59:59Z" + + def test_to_params_with_to_and_sending_domain_id(self) -> None: + f = EmailLogsListFilters( + to=filter_ci_equal("recipient@example.com"), + sending_domain_id=filter_sending_domain_id_equal([3938, 3939]), + ) + params = f.to_params() + assert params["filters[to][operator]"] == "ci_equal" + assert params["filters[to][value]"] == "recipient@example.com" + assert params["filters[sending_domain_id][operator]"] == "equal" + assert params["filters[sending_domain_id][value][]"] == [3938, 3939]