diff --git a/README.md b/README.md index d548f8f..9dc76d2 100644 --- a/README.md +++ b/README.md @@ -241,6 +241,9 @@ The same situation applies to both `client.batch_send()` and `client.sending_api ### Email Templates API: - Templates management – [`email_templates/templates.py`](examples/email_templates/templates.py) +### Sending Domains API: +- Sending Domains – [`sending_domains/sending_domains.py`](examples/sending_domains/sending_domains.py) + ### Suppressions API: - Suppressions (find & delete) – [`suppressions/suppressions.py`](examples/suppressions/suppressions.py) diff --git a/examples/sending_domains/sending_domains.py b/examples/sending_domains/sending_domains.py new file mode 100644 index 0000000..d988696 --- /dev/null +++ b/examples/sending_domains/sending_domains.py @@ -0,0 +1,49 @@ +import mailtrap as mt +from mailtrap.models.common import DeletedObject +from mailtrap.models.sending_domains import SendingDomain +from mailtrap.models.sending_domains import SendSetupInstructionsResponse + +API_TOKEN = "YOUR_API_TOKEN" +ACCOUNT_ID = "YOUR_ACCOUNT_ID" + +client = mt.MailtrapClient(token=API_TOKEN, account_id=ACCOUNT_ID) +sending_domains_api = client.sending_domains_api.sending_domains + + +def list_sending_domains() -> list[SendingDomain]: + return sending_domains_api.get_list() + + +def get_sending_domain(domain_id: int) -> SendingDomain: + return sending_domains_api.get_by_id(domain_id) + + +def create_sending_domain(domain_name: str) -> SendingDomain: + params = mt.CreateSendingDomainParams(domain_name=domain_name) + return sending_domains_api.create(params) + + +def delete_sending_domain(domain_id: int) -> DeletedObject: + return sending_domains_api.delete(domain_id) + + +def send_setup_instructions(domain_id: int, email: str) -> SendSetupInstructionsResponse: + params = mt.SendSetupInstructionsParams(email=email) + return sending_domains_api.send_setup_instructions(domain_id, params) + + +if __name__ == "__main__": + new_domain = create_sending_domain("example.com") + print(new_domain) + + domains = list_sending_domains() + print(domains) + + domain = get_sending_domain(new_domain.id) + print(domain) + + response = send_setup_instructions(new_domain.id, "example@mail.com") + print(response) + + deleted_domain = delete_sending_domain(new_domain.id) + print(deleted_domain) diff --git a/mailtrap/__init__.py b/mailtrap/__init__.py index 999a1a1..67b5832 100644 --- a/mailtrap/__init__.py +++ b/mailtrap/__init__.py @@ -30,5 +30,7 @@ from .models.messages import UpdateEmailMessageParams from .models.permissions import PermissionResourceParams from .models.projects import ProjectParams +from .models.sending_domains import CreateSendingDomainParams +from .models.sending_domains import SendSetupInstructionsParams from .models.templates import CreateEmailTemplateParams from .models.templates import UpdateEmailTemplateParams diff --git a/mailtrap/api/resources/sending_domains.py b/mailtrap/api/resources/sending_domains.py new file mode 100644 index 0000000..c7982ba --- /dev/null +++ b/mailtrap/api/resources/sending_domains.py @@ -0,0 +1,69 @@ +from typing import Optional + +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.sending_domains import CreateSendingDomainParams +from mailtrap.models.sending_domains import SendingDomain +from mailtrap.models.sending_domains import SendSetupInstructionsParams +from mailtrap.models.sending_domains import SendSetupInstructionsResponse + + +class SendingDomainsApi: + def __init__(self, client: HttpClient, account_id: str) -> None: + self._account_id = account_id + self._client = client + + def get_list(self) -> list[SendingDomain]: + """ + Get sending domains and their statuses. + """ + response = self._client.get(self._api_path()) + domains = response.get("data", []) + return [SendingDomain(**domain) for domain in domains] + + def get_by_id(self, sending_domain_id: int) -> SendingDomain: + """ + Get domain data and its status. + """ + response = self._client.get(self._api_path(sending_domain_id)) + return SendingDomain(**response) + + def create(self, domain_params: CreateSendingDomainParams) -> SendingDomain: + """ + Create a sending domain. To later check the status of the newly created domain, + review the compliance_status and dns_verified fields in the response + of the Get domain by ID or Get sending domains endpoints. + """ + response = self._client.post( + self._api_path(), json={"sending_domain": domain_params.api_data} + ) + return SendingDomain(**response) + + def delete(self, sending_domain_id: int) -> DeletedObject: + """ + Delete a sending domain. + """ + self._client.delete(self._api_path(sending_domain_id)) + return DeletedObject(id=sending_domain_id) + + def send_setup_instructions( + self, + sending_domain_id: int, + instructions_params: SendSetupInstructionsParams, + ) -> SendSetupInstructionsResponse: + """ + Send sending domain setup instructions. + """ + self._client.post( + f"{self._api_path(sending_domain_id)}/send_setup_instructions", + json=instructions_params.api_data, + ) + return SendSetupInstructionsResponse( + message="Instructions email has been sent successfully" + ) + + def _api_path(self, sending_domain_id: Optional[int] = None) -> str: + path = f"/api/accounts/{self._account_id}/sending_domains" + if sending_domain_id is not None: + path = f"{path}/{sending_domain_id}" + return path diff --git a/mailtrap/api/sending_domains.py b/mailtrap/api/sending_domains.py new file mode 100644 index 0000000..b12b107 --- /dev/null +++ b/mailtrap/api/sending_domains.py @@ -0,0 +1,12 @@ +from mailtrap.api.resources.sending_domains import SendingDomainsApi +from mailtrap.http import HttpClient + + +class SendingDomainsBaseApi: + def __init__(self, client: HttpClient, account_id: str) -> None: + self._account_id = account_id + self._client = client + + @property + def sending_domains(self) -> SendingDomainsApi: + return SendingDomainsApi(account_id=self._account_id, client=self._client) diff --git a/mailtrap/client.py b/mailtrap/client.py index fb97a43..12fc933 100644 --- a/mailtrap/client.py +++ b/mailtrap/client.py @@ -8,6 +8,7 @@ from mailtrap.api.contacts import ContactsBaseApi from mailtrap.api.general import GeneralApi from mailtrap.api.sending import SendingApi +from mailtrap.api.sending_domains import SendingDomainsBaseApi from mailtrap.api.suppressions import SuppressionsBaseApi from mailtrap.api.templates import EmailTemplatesApi from mailtrap.api.testing import TestingApi @@ -93,6 +94,14 @@ def suppressions_api(self) -> SuppressionsBaseApi: client=HttpClient(host=GENERAL_HOST, headers=self.headers), ) + @property + def sending_domains_api(self) -> SendingDomainsBaseApi: + self._validate_account_id() + return SendingDomainsBaseApi( + 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) diff --git a/mailtrap/models/sending_domains.py b/mailtrap/models/sending_domains.py new file mode 100644 index 0000000..a470c0e --- /dev/null +++ b/mailtrap/models/sending_domains.py @@ -0,0 +1,57 @@ +from typing import Optional + +from pydantic import Field +from pydantic.dataclasses import dataclass + +from mailtrap.models.common import RequestParams + + +@dataclass +class SendingDomainPermissions: + can_read: bool + can_update: bool + can_destroy: bool + + +@dataclass +class DnsRecord: + key: str + domain: str + type: str + value: str + status: str + name: str + + +@dataclass +class SendingDomain: + id: int + domain_name: str + demo: bool + compliance_status: str + dns_verified: bool + open_tracking_enabled: bool + click_tracking_enabled: bool + auto_unsubscribe_link_enabled: bool + custom_domain_tracking_enabled: bool + health_alerts_enabled: bool + critical_alerts_enabled: bool + permissions: SendingDomainPermissions + alert_recipient_email: Optional[str] = None + dns_verified_at: Optional[str] = None + dns_records: list[DnsRecord] = Field(default_factory=list) + + +@dataclass +class CreateSendingDomainParams(RequestParams): + domain_name: str + + +@dataclass +class SendSetupInstructionsParams(RequestParams): + email: str + + +@dataclass +class SendSetupInstructionsResponse: + message: str diff --git a/tests/unit/api/sending_domains/__init__.py b/tests/unit/api/sending_domains/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/api/sending_domains/test_sending_domains.py b/tests/unit/api/sending_domains/test_sending_domains.py new file mode 100644 index 0000000..3c810ab --- /dev/null +++ b/tests/unit/api/sending_domains/test_sending_domains.py @@ -0,0 +1,346 @@ +from typing import Any + +import pytest +import responses + +from mailtrap.api.resources.sending_domains import SendingDomainsApi +from mailtrap.config import GENERAL_HOST +from mailtrap.exceptions import APIError +from mailtrap.http import HttpClient +from mailtrap.models.common import DeletedObject +from mailtrap.models.sending_domains import CreateSendingDomainParams +from mailtrap.models.sending_domains import SendingDomain +from mailtrap.models.sending_domains import SendSetupInstructionsParams +from mailtrap.models.sending_domains import SendSetupInstructionsResponse +from tests import conftest + +ACCOUNT_ID = "1234567" +DOMAIN_ID = 432 +BASE_SENDING_DOMAINS_URL = ( + f"https://{GENERAL_HOST}/api/accounts/{ACCOUNT_ID}/sending_domains" +) + + +@pytest.fixture +def sending_domains_api() -> SendingDomainsApi: + return SendingDomainsApi(account_id=ACCOUNT_ID, client=HttpClient(GENERAL_HOST)) + + +@pytest.fixture +def sample_sending_domain_dict() -> dict[str, Any]: + return { + "id": DOMAIN_ID, + "domain_name": "example.com", + "demo": False, + "compliance_status": "verified", + "dns_verified": True, + "open_tracking_enabled": True, + "click_tracking_enabled": True, + "auto_unsubscribe_link_enabled": False, + "custom_domain_tracking_enabled": False, + "health_alerts_enabled": True, + "critical_alerts_enabled": True, + "permissions": { + "can_read": True, + "can_update": True, + "can_destroy": True, + }, + "alert_recipient_email": "admin@example.com", + "dns_verified_at": "2024-12-26T09:40:44.161Z", + "dns_records": [ + { + "key": "dkim", + "domain": "example.com", + "type": "CNAME", + "value": "example.com.mailtrap.io", + "status": "verified", + "name": "mail", + } + ], + } + + +class TestSendingDomainsApi: + + @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_sending_domains_should_raise_api_errors( + self, + sending_domains_api: SendingDomainsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get( + BASE_SENDING_DOMAINS_URL, + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + sending_domains_api.get_list() + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_get_sending_domains_should_return_domains_list( + self, sending_domains_api: SendingDomainsApi, sample_sending_domain_dict: dict + ) -> None: + responses.get( + BASE_SENDING_DOMAINS_URL, + json={"data": [sample_sending_domain_dict]}, + status=200, + ) + + domains = sending_domains_api.get_list() + + assert isinstance(domains, list) + assert all(isinstance(d, SendingDomain) for d in domains) + assert domains[0].id == DOMAIN_ID + assert domains[0].domain_name == "example.com" + assert domains[0].dns_verified is True + + @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_get_sending_domain_by_id_should_raise_api_errors( + self, + sending_domains_api: SendingDomainsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.get( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}", + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + sending_domains_api.get_by_id(DOMAIN_ID) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_get_sending_domain_by_id_should_return_sending_domain( + self, sending_domains_api: SendingDomainsApi, sample_sending_domain_dict: dict + ) -> None: + responses.get( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}", + json=sample_sending_domain_dict, + status=200, + ) + + domain = sending_domains_api.get_by_id(DOMAIN_ID) + + assert isinstance(domain, SendingDomain) + assert domain.id == DOMAIN_ID + assert domain.domain_name == "example.com" + assert domain.compliance_status == "verified" + assert domain.dns_verified is True + + @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.VALIDATION_ERRORS_STATUS_CODE, + { + "errors": { + "base": ["Validation failed: Domain name has already been taken"] + } + }, + "Validation failed", + ), + ], + ) + @responses.activate + def test_create_sending_domain_should_raise_api_errors( + self, + sending_domains_api: SendingDomainsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.post( + BASE_SENDING_DOMAINS_URL, + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + params = CreateSendingDomainParams(domain_name="example.com") + sending_domains_api.create(params) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_create_sending_domain_should_return_created_domain( + self, sending_domains_api: SendingDomainsApi, sample_sending_domain_dict: dict + ) -> None: + responses.post( + BASE_SENDING_DOMAINS_URL, + json=sample_sending_domain_dict, + status=200, + ) + + params = CreateSendingDomainParams(domain_name="example.com") + domain = sending_domains_api.create(params) + + assert isinstance(domain, SendingDomain) + assert domain.id == DOMAIN_ID + assert domain.domain_name == "example.com" + + @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.INTERNAL_SERVER_ERROR_STATUS_CODE, + conftest.INTERNAL_SERVER_ERROR_RESPONSE, + conftest.INTERNAL_SERVER_ERROR_MESSAGE, + ), + ], + ) + @responses.activate + def test_delete_sending_domain_should_raise_api_errors( + self, + sending_domains_api: SendingDomainsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.delete( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}", + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + sending_domains_api.delete(DOMAIN_ID) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_delete_sending_domain_should_return_deleted_object( + self, sending_domains_api: SendingDomainsApi + ) -> None: + responses.delete( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}", + status=204, + ) + + deleted_domain = sending_domains_api.delete(DOMAIN_ID) + + assert isinstance(deleted_domain, DeletedObject) + assert deleted_domain.id == DOMAIN_ID + + @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_send_setup_instructions_should_raise_api_errors( + self, + sending_domains_api: SendingDomainsApi, + status_code: int, + response_json: dict, + expected_error_message: str, + ) -> None: + responses.post( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}/send_setup_instructions", + status=status_code, + json=response_json, + ) + + with pytest.raises(APIError) as exc_info: + params = SendSetupInstructionsParams(email="admin@example.com") + sending_domains_api.send_setup_instructions(DOMAIN_ID, params) + + assert expected_error_message in str(exc_info.value) + + @responses.activate + def test_send_setup_instructions_should_return_success_response( + self, sending_domains_api: SendingDomainsApi + ) -> None: + responses.post( + f"{BASE_SENDING_DOMAINS_URL}/{DOMAIN_ID}/send_setup_instructions", + status=204, + ) + + params = SendSetupInstructionsParams(email="admin@example.com") + response = sending_domains_api.send_setup_instructions(DOMAIN_ID, params) + + assert isinstance(response, SendSetupInstructionsResponse) + assert response.message == "Instructions email has been sent successfully" diff --git a/tests/unit/models/test_sending_domains.py b/tests/unit/models/test_sending_domains.py new file mode 100644 index 0000000..43d7160 --- /dev/null +++ b/tests/unit/models/test_sending_domains.py @@ -0,0 +1,14 @@ +from mailtrap.models.sending_domains import CreateSendingDomainParams +from mailtrap.models.sending_domains import SendSetupInstructionsParams + + +class TestCreateSendingDomainParams: + def test_api_data_should_return_dict_with_all_props(self) -> None: + entity = CreateSendingDomainParams(domain_name="test.co") + assert entity.api_data == {"domain_name": "test.co"} + + +class TestSendSetupInstructionsParams: + def test_api_data_should_return_dict_with_all_props(self) -> None: + entity = SendSetupInstructionsParams(email="example@mail.com") + assert entity.api_data == {"email": "example@mail.com"}