diff --git a/examples/terminal_group_manager/get_terminals_by_group.py b/examples/terminal_group_manager/get_terminals_by_group.py new file mode 100644 index 0000000..fb6a9ae --- /dev/null +++ b/examples/terminal_group_manager/get_terminals_by_group.py @@ -0,0 +1,57 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +import os + +from dotenv import load_dotenv + +from multisafepay import Sdk +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + +default_account_api_key = (os.getenv("API_KEY") or "").strip() +partner_affiliate_api_key = (os.getenv("PARTNER_API_KEY") or "").strip() + +terminal_group_id = os.getenv( + "CLOUD_POS_TERMINAL_GROUP_ID", + "", +).strip() + +if __name__ == "__main__": + # get_terminals_by_group → partner_affiliate scope → resolver returns + # partner_affiliate_api_key, falls back to default_api_key + resolver_kwargs = { + "default_api_key": default_account_api_key, + "partner_affiliate_api_key": partner_affiliate_api_key, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + + # Get the 'TerminalGroup' manager from the SDK + terminal_group_manager = multisafepay_sdk.get_terminal_group_manager() + + # Define optional pagination parameters + options = { + "limit": 10, + "page": 1, + } + + # Fetch terminals assigned to the specified terminal group + terminals_by_group_response = terminal_group_manager.get_terminals_by_group( + terminal_group_id=terminal_group_id, + options=options, + ) + + # Print the terminal listing data + print(terminals_by_group_response.get_data()) diff --git a/examples/terminal_manager/create.py b/examples/terminal_manager/create.py new file mode 100644 index 0000000..9c9354c --- /dev/null +++ b/examples/terminal_manager/create.py @@ -0,0 +1,61 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +import os + +from dotenv import load_dotenv + +from multisafepay import Sdk +from multisafepay.api.paths.terminals.request.create_terminal_request import ( + CreateTerminalRequest, +) +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + +default_account_api_key = (os.getenv("API_KEY") or "").strip() +partner_affiliate_api_key = (os.getenv("PARTNER_API_KEY") or "").strip() + +terminal_group_id_raw = os.getenv( + "CLOUD_POS_TERMINAL_GROUP_ID", + "", +).strip() + +if __name__ == "__main__": + # create_terminal → default scope → resolver returns default_api_key + terminal_group_id = int(terminal_group_id_raw) + + resolver_kwargs = { + "default_api_key": default_account_api_key, + "partner_affiliate_api_key": partner_affiliate_api_key, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + + # Get the 'Terminal' manager from the SDK + terminal_manager = multisafepay_sdk.get_terminal_manager() + + # Build the create terminal request + create_request = ( + CreateTerminalRequest() + .add_provider("CTAP") + .add_group_id(terminal_group_id) + .add_name("Demo POS Terminal") + ) + + # Create a new POS terminal + terminal_response = terminal_manager.create_terminal(create_request) + + # Print the created terminal data + terminal_data = terminal_response.get_data() + print(terminal_data) diff --git a/examples/terminal_manager/get_terminals.py b/examples/terminal_manager/get_terminals.py new file mode 100644 index 0000000..31fa310 --- /dev/null +++ b/examples/terminal_manager/get_terminals.py @@ -0,0 +1,49 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +import os + +from dotenv import load_dotenv + +from multisafepay import Sdk +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + +default_account_api_key = (os.getenv("API_KEY") or "").strip() +partner_affiliate_api_key = (os.getenv("PARTNER_API_KEY") or "").strip() + +if __name__ == "__main__": + # get_terminals → partner_affiliate scope → resolver returns + # partner_affiliate_api_key, falls back to default_api_key + resolver_kwargs = { + "default_api_key": default_account_api_key, + "partner_affiliate_api_key": partner_affiliate_api_key, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + + # Get the 'Terminal' manager from the SDK + terminal_manager = multisafepay_sdk.get_terminal_manager() + + # Define optional pagination parameters + options = { + "limit": 10, + "page": 1, + } + + # Fetch terminals for the account + terminals_response = terminal_manager.get_terminals(options=options) + + # Print the terminal listing data + print(terminals_response.get_data()) diff --git a/src/multisafepay/api/paths/terminal_groups/__init__.py b/src/multisafepay/api/paths/terminal_groups/__init__.py new file mode 100644 index 0000000..22fc7e5 --- /dev/null +++ b/src/multisafepay/api/paths/terminal_groups/__init__.py @@ -0,0 +1,16 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Terminal group API endpoints.""" + +from multisafepay.api.paths.terminal_groups.terminal_group_manager import ( + TerminalGroupManager, +) + +__all__ = [ + "TerminalGroupManager", +] diff --git a/src/multisafepay/api/paths/terminal_groups/terminal_group_manager.py b/src/multisafepay/api/paths/terminal_groups/terminal_group_manager.py new file mode 100644 index 0000000..e1ec30e --- /dev/null +++ b/src/multisafepay/api/paths/terminal_groups/terminal_group_manager.py @@ -0,0 +1,107 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Terminal group manager for `/json/terminal-groups/{terminal_group_id}/terminals`.""" + +from multisafepay.api.base.abstract_manager import AbstractManager +from multisafepay.api.base.listings.listing_pager import ListingPager +from multisafepay.api.base.listings.pager import Pager +from multisafepay.api.base.response.api_response import ApiResponse +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminals.response.terminal import Terminal +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope +from multisafepay.util.message import MessageList, gen_could_not_created_msg +from pydantic import ValidationError + +ALLOWED_OPTIONS = { + "page": "", + "limit": "", +} + + +class TerminalGroupManager(AbstractManager): + """A class representing the TerminalGroupManager.""" + + def __init__(self: "TerminalGroupManager", client: Client) -> None: + """ + Initialize the TerminalGroupManager with a client. + + Parameters + ---------- + client (Client): The client used to make API requests. + + """ + super().__init__(client) + + @staticmethod + def __custom_terminal_listing_response( + response: ApiResponse, + ) -> CustomApiResponse: + args: dict = { + **response.dict(), + "data": None, + } + + pager = None + raw_pager = response.get_pager() + if isinstance(raw_pager, dict): + pager = Pager.from_dict(raw_pager.copy()) + + try: + args["data"] = ListingPager( + data=response.get_body_data().copy(), + pager=pager, + class_type=Terminal, + ) + except (AttributeError, TypeError, ValidationError): + args["warnings"] = MessageList().add_message( + gen_could_not_created_msg("Listing Terminal"), + ) + + return CustomApiResponse(**args) + + def get_terminals_by_group( + self: "TerminalGroupManager", + terminal_group_id: str, + options: dict = None, + ) -> CustomApiResponse: + """ + List POS terminals for the given terminal group. + + Parameters + ---------- + terminal_group_id (str): Terminal group identifier. + options (dict): Request options (`page`, `limit`). Defaults to None. + + Returns + ------- + CustomApiResponse: The response containing terminal listing data. + + """ + if options is None: + options = {} + options = {k: v for k, v in options.items() if k in ALLOWED_OPTIONS} + + encoded_terminal_group_id = self.encode_path_segment(terminal_group_id) + endpoint = ( + f"json/terminal-groups/{encoded_terminal_group_id}/terminals" + ) + context = {"terminal_group_id": terminal_group_id} + response = self.client.create_get_request( + endpoint=endpoint, + params=options, + context=context, + auth_scope=AuthScope( + scope=Client.AUTH_SCOPE_PARTNER_AFFILIATE, + ), + ) + return TerminalGroupManager.__custom_terminal_listing_response( + response, + ) diff --git a/src/multisafepay/api/paths/terminals/__init__.py b/src/multisafepay/api/paths/terminals/__init__.py new file mode 100644 index 0000000..1690144 --- /dev/null +++ b/src/multisafepay/api/paths/terminals/__init__.py @@ -0,0 +1,16 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Terminal API endpoints for POS terminal and receipt operations.""" + +from multisafepay.api.paths.terminals.terminal_manager import ( + TerminalManager, +) + +__all__ = [ + "TerminalManager", +] diff --git a/src/multisafepay/api/paths/terminals/request/__init__.py b/src/multisafepay/api/paths/terminals/request/__init__.py new file mode 100644 index 0000000..5b57ad5 --- /dev/null +++ b/src/multisafepay/api/paths/terminals/request/__init__.py @@ -0,0 +1,16 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Request models for terminal-related API calls.""" + +from multisafepay.api.paths.terminals.request.create_terminal_request import ( + CreateTerminalRequest, +) + +__all__ = [ + "CreateTerminalRequest", +] diff --git a/src/multisafepay/api/paths/terminals/request/create_terminal_request.py b/src/multisafepay/api/paths/terminals/request/create_terminal_request.py new file mode 100644 index 0000000..69cbe0b --- /dev/null +++ b/src/multisafepay/api/paths/terminals/request/create_terminal_request.py @@ -0,0 +1,103 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Request model for creating POS terminals.""" + +from typing import Optional + +from multisafepay.exception.invalid_argument import InvalidArgumentException +from multisafepay.model.request_model import RequestModel + +CTAP_PROVIDER = "CTAP" +ALLOWED_PROVIDERS = [ + CTAP_PROVIDER, +] + + +class CreateTerminalRequest(RequestModel): + """ + Request body for the create terminal endpoint. + + Attributes + ---------- + provider (Optional[str]): The terminal provider. + group_id (Optional[int]): The terminal group id. + name (Optional[str]): The terminal name. + + """ + + provider: Optional[str] = None + group_id: Optional[int] = None + name: Optional[str] = None + + def add_provider( + self: "CreateTerminalRequest", + provider: Optional[str], + ) -> "CreateTerminalRequest": + """ + Add a terminal provider. + + Parameters + ---------- + provider (Optional[str]): The provider value. + + Raises + ------ + InvalidArgumentException: If provider is not one of the allowed values. + + Returns + ------- + CreateTerminalRequest: The current request object. + + """ + if provider is not None and provider not in ALLOWED_PROVIDERS: + msg = ( + f'Provider "{provider}" is not a known provider. ' + f'Available providers: {", ".join(ALLOWED_PROVIDERS)}' + ) + raise InvalidArgumentException(msg) + + self.provider = provider + return self + + def add_group_id( + self: "CreateTerminalRequest", + group_id: str, + ) -> "CreateTerminalRequest": + """ + Add a terminal group id. + + Parameters + ---------- + group_id (str): The terminal group identifier. + + Returns + ------- + CreateTerminalRequest: The current request object. + + """ + self.group_id = group_id + return self + + def add_name( + self: "CreateTerminalRequest", + name: Optional[str], + ) -> "CreateTerminalRequest": + """ + Add a terminal name. + + Parameters + ---------- + name (Optional[str]): The terminal name. + + Returns + ------- + CreateTerminalRequest: The current request object. + + """ + self.name = name + return self diff --git a/src/multisafepay/api/paths/terminals/response/__init__.py b/src/multisafepay/api/paths/terminals/response/__init__.py new file mode 100644 index 0000000..6dc8ae3 --- /dev/null +++ b/src/multisafepay/api/paths/terminals/response/__init__.py @@ -0,0 +1,14 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Response models for terminal endpoints.""" + +from multisafepay.api.paths.terminals.response.terminal import Terminal + +__all__ = [ + "Terminal", +] diff --git a/src/multisafepay/api/paths/terminals/response/terminal.py b/src/multisafepay/api/paths/terminals/response/terminal.py new file mode 100644 index 0000000..45e28f3 --- /dev/null +++ b/src/multisafepay/api/paths/terminals/response/terminal.py @@ -0,0 +1,63 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Response model for POS terminal data.""" + +from typing import Optional + +from multisafepay.model.response_model import ResponseModel + + +class Terminal(ResponseModel): + """ + Represents a POS terminal returned by the API. + + Attributes + ---------- + id (Optional[str]): The terminal identifier. + provider (Optional[str]): The terminal provider. + name (Optional[str]): The terminal name. + code (Optional[str]): The terminal code. + created (Optional[str]): Terminal creation timestamp. + last_updated (Optional[str]): Terminal update timestamp. + manufacturer_id (Optional[str]): Terminal manufacturer identifier. + serial_number (Optional[str]): Terminal serial number. + active (Optional[bool]): Whether the terminal is active. + group_id (Optional[int]): The terminal group identifier. + country (Optional[str]): The terminal country code. + + """ + + id: Optional[str] + provider: Optional[str] + name: Optional[str] + code: Optional[str] + created: Optional[str] + last_updated: Optional[str] + manufacturer_id: Optional[str] + serial_number: Optional[str] + active: Optional[bool] + group_id: Optional[int] + country: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["Terminal"]: + """ + Create a Terminal from dictionary data. + + Parameters + ---------- + d (dict): The terminal data. + + Returns + ------- + Optional[Terminal]: A terminal instance or None. + + """ + if d is None: + return None + return Terminal(**d) diff --git a/src/multisafepay/api/paths/terminals/terminal_manager.py b/src/multisafepay/api/paths/terminals/terminal_manager.py new file mode 100644 index 0000000..41a5930 --- /dev/null +++ b/src/multisafepay/api/paths/terminals/terminal_manager.py @@ -0,0 +1,144 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Terminal manager for `/json/terminals` operations.""" + +import json + +from multisafepay.api.base.abstract_manager import AbstractManager +from multisafepay.api.base.listings.listing_pager import ListingPager +from multisafepay.api.base.listings.pager import Pager +from multisafepay.api.base.response.api_response import ApiResponse +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminals.request.create_terminal_request import ( + CreateTerminalRequest, +) +from multisafepay.api.paths.terminals.response.terminal import Terminal +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope +from multisafepay.util.dict_utils import dict_empty +from multisafepay.util.message import MessageList, gen_could_not_created_msg +from pydantic import ValidationError + +ALLOWED_OPTIONS = { + "page": "", + "limit": "", +} + + +class TerminalManager(AbstractManager): + """A class representing the TerminalManager.""" + + def __init__(self: "TerminalManager", client: Client) -> None: + """ + Initialize the TerminalManager with a client. + + Parameters + ---------- + client (Client): The client used to make API requests. + + """ + super().__init__(client) + + @staticmethod + def __custom_terminal_response(response: ApiResponse) -> CustomApiResponse: + args: dict = { + **response.dict(), + "data": None, + } + if not dict_empty(response.get_body_data()): + try: + args["data"] = Terminal.from_dict( + d=response.get_body_data().copy(), + ) + except ValidationError: + args["warnings"] = MessageList().add_message( + gen_could_not_created_msg("Terminal"), + ) + + return CustomApiResponse(**args) + + @staticmethod + def __custom_terminal_listing_response( + response: ApiResponse, + ) -> CustomApiResponse: + args: dict = { + **response.dict(), + "data": None, + } + + pager = None + raw_pager = response.get_pager() + if isinstance(raw_pager, dict): + pager = Pager.from_dict(raw_pager.copy()) + + try: + args["data"] = ListingPager( + data=response.get_body_data().copy(), + pager=pager, + class_type=Terminal, + ) + except (AttributeError, TypeError, ValidationError): + args["warnings"] = MessageList().add_message( + gen_could_not_created_msg("Listing Terminal"), + ) + + return CustomApiResponse(**args) + + def create_terminal( + self: "TerminalManager", + create_terminal_request: CreateTerminalRequest, + ) -> CustomApiResponse: + """ + Create a new POS terminal. + + Parameters + ---------- + create_terminal_request (CreateTerminalRequest): Request payload. + + Returns + ------- + CustomApiResponse: The response containing created terminal data. + + """ + json_data = json.dumps(create_terminal_request.to_dict()) + response = self.client.create_post_request( + "json/terminals", + request_body=json_data, + ) + return TerminalManager.__custom_terminal_response(response) + + def get_terminals( + self: "TerminalManager", + options: dict = None, + ) -> CustomApiResponse: + """ + List POS terminals for the account. + + Parameters + ---------- + options (dict): Request options (`page`, `limit`). Defaults to None. + + Returns + ------- + CustomApiResponse: The response containing terminal listing data. + + """ + if options is None: + options = {} + options = {k: v for k, v in options.items() if k in ALLOWED_OPTIONS} + + response = self.client.create_get_request( + "json/terminals", + options, + auth_scope=AuthScope( + scope=Client.AUTH_SCOPE_PARTNER_AFFILIATE, + ), + ) + return TerminalManager.__custom_terminal_listing_response(response) diff --git a/src/multisafepay/sdk.py b/src/multisafepay/sdk.py index 818315b..e99eb16 100644 --- a/src/multisafepay/sdk.py +++ b/src/multisafepay/sdk.py @@ -17,6 +17,10 @@ from multisafepay.api.paths.payment_methods.payment_method_manager import ( PaymentMethodManager, ) +from multisafepay.api.paths.terminal_groups.terminal_group_manager import ( + TerminalGroupManager, +) +from multisafepay.api.paths.terminals.terminal_manager import TerminalManager from multisafepay.api.paths.transactions.transaction_manager import ( TransactionManager, ) @@ -197,6 +201,30 @@ def get_capture_manager(self: "Sdk") -> CaptureManager: """ return CaptureManager(self.client) + def get_terminal_manager(self: "Sdk") -> TerminalManager: + """ + Get the terminal manager. + + Returns + ------- + TerminalManager + The terminal manager instance. + + """ + return TerminalManager(self.client) + + def get_terminal_group_manager(self: "Sdk") -> TerminalGroupManager: + """ + Get the terminal group manager. + + Returns + ------- + TerminalGroupManager + The terminal group manager instance. + + """ + return TerminalGroupManager(self.client) + def get_client(self: "Sdk") -> Client: """ Get the client instance. diff --git a/tests/multisafepay/e2e/conftest.py b/tests/multisafepay/e2e/conftest.py index 66d1a96..1f74697 100644 --- a/tests/multisafepay/e2e/conftest.py +++ b/tests/multisafepay/e2e/conftest.py @@ -28,10 +28,10 @@ def _get_e2e_base_url() -> str: return base_url or Client.TEST_URL -def _validate_e2e_base_url(base_url: str) -> str: +def _validate_e2e_base_url(base_url: str, env_name: str) -> str: parsed = urlparse(base_url) if parsed.scheme != "https" or not parsed.netloc: - msg = f"{E2E_BASE_URL_ENV} must be a valid https URL" + msg = f"{env_name} must be a valid https URL" raise pytest.UsageError(msg) parsed = urlparse(base_url) @@ -53,7 +53,7 @@ def e2e_api_key() -> str: @pytest.fixture(scope="session") def e2e_base_url() -> str: """Return the dedicated base URL used by E2E tests.""" - return _validate_e2e_base_url(_get_e2e_base_url()) + return _validate_e2e_base_url(_get_e2e_base_url(), E2E_BASE_URL_ENV) @pytest.fixture(scope="session") @@ -80,27 +80,3 @@ def create_sdk(*, transport: Optional[HTTPTransport] = None) -> Sdk: def e2e_sdk(e2e_sdk_factory: Callable[..., Sdk]) -> Sdk: """Return the default SDK instance used by E2E tests.""" return e2e_sdk_factory() - - -def pytest_collection_modifyitems( - config: pytest.Config, # noqa: ARG001 - items: list[pytest.Item], -) -> None: - """ - Skip all e2e tests when E2E_API_KEY is missing. - - These tests perform real API calls. In most local/CI environments the secret - isn't present, so we prefer a clean skip over hard errors during fixture setup. - """ - if _get_e2e_api_key(): - return - - skip = pytest.mark.skip( - reason=f"E2E tests require {E2E_API_KEY_ENV} (not set)", - ) - for item in items: - # This hook runs for the whole session (all collected tests), even when - # this conftest is only loaded due to e2e tests being present/deselected. - # Ensure we only affect e2e tests. - if item.nodeid.startswith("tests/multisafepay/e2e/"): - item.add_marker(skip) diff --git a/tests/multisafepay/e2e/examples/conftest.py b/tests/multisafepay/e2e/examples/conftest.py new file mode 100644 index 0000000..718339f --- /dev/null +++ b/tests/multisafepay/e2e/examples/conftest.py @@ -0,0 +1,206 @@ +"""Example-specific E2E fixtures and selective skip behavior.""" + +import os +from typing import Optional +from urllib.parse import urlparse + +import pytest + +from multisafepay.client import ScopedCredentialResolver +from multisafepay.sdk import Sdk + +DEFAULT_E2E_API_KEY_ENV = "E2E_API_KEY" +TERMINAL_DEFAULT_API_KEY_ENV = "API_KEY" +TERMINAL_PARTNER_API_KEY_ENV = "PARTNER_API_KEY" +TERMINAL_CUSTOM_BASE_URL_ENV = "MSP_SDK_CUSTOM_BASE_URL" +TERMINAL_E2E_TERMINAL_ID_ENV = "E2E_CLOUD_POS_TERMINAL_ID" +TERMINAL_E2E_NODE_PREFIXES = ( + "tests/multisafepay/e2e/examples/terminal_manager/", + "tests/multisafepay/e2e/examples/terminal_group_manager/", +) + + +def _get_first_env(*names: str) -> str: + for name in names: + value = os.getenv(name, "").strip() + if value: + return value + + return "" + + +def _require_env(name: str) -> str: + value = os.getenv(name, "").strip() + if not value: + msg = f"Feature-specific E2E tests require {name} (not set)" + raise pytest.UsageError(msg) + return value + + +def _has_terminal_e2e_env() -> bool: + return bool( + _get_first_env(TERMINAL_DEFAULT_API_KEY_ENV) + and _get_first_env(TERMINAL_CUSTOM_BASE_URL_ENV) + and _get_first_env(TERMINAL_E2E_TERMINAL_ID_ENV), + ) + + +def _validate_base_url(base_url: str, env_name: str) -> str: + parsed = urlparse(base_url) + if parsed.scheme != "https" or not parsed.netloc: + msg = f"{env_name} must be a valid https URL" + raise pytest.UsageError(msg) + + path = parsed.path.rstrip("/") + normalized_path = "/" if not path else f"{path}/" + return f"{parsed.scheme}://{parsed.netloc}{normalized_path}" + + +def _resolve_terminal_group_id( + terminals_sdk: Sdk, + terminal_id: str, + label: str, +) -> str: + terminal_manager = terminals_sdk.get_terminal_manager() + limit = 100 + max_pages = 10 + + for page in range(1, max_pages + 1): + response = terminal_manager.get_terminals( + options={ + "limit": limit, + "page": page, + }, + ) + if ( + response.get_status_code() != 200 + or not response.get_body_success() + ): + raise pytest.UsageError( + "Unable to resolve terminal group id: " + "GET /json/terminals did not return a successful response", + ) + + listing = response.get_data() + if listing is None: + break + + terminals = listing.get_data() + for terminal in terminals: + listed_terminal_id = getattr(terminal, "id", None) + terminal_code = getattr(terminal, "code", None) + if terminal_id not in {listed_terminal_id, terminal_code}: + continue + + group_id = getattr(terminal, "group_id", None) + if group_id is None: + raise pytest.UsageError( + f"Unable to resolve {label}: " + f"terminal {terminal_id} has no group_id", + ) + return str(group_id) + + if len(terminals) < limit: + break + + raise pytest.UsageError( + f"Unable to resolve {label} from /json/terminals " + f"for terminal {terminal_id}", + ) + + +@pytest.fixture(scope="session") +def terminals_terminal_id() -> str: + """Return terminal id used to resolve a valid terminal group.""" + return _require_env(TERMINAL_E2E_TERMINAL_ID_ENV) + + +@pytest.fixture(scope="session") +def terminals_e2e_api_key() -> str: + """Return default API key used by terminal endpoint E2E tests.""" + return _require_env(TERMINAL_DEFAULT_API_KEY_ENV) + + +@pytest.fixture(scope="session") +def terminals_partner_affiliate_api_key() -> Optional[str]: + """Return partner key for terminal endpoint E2E tests when available.""" + api_key = _get_first_env(TERMINAL_PARTNER_API_KEY_ENV) + return api_key or None + + +@pytest.fixture(scope="session") +def terminals_e2e_base_url() -> str: + """Return custom base URL used by terminal endpoint E2E tests.""" + return _validate_base_url( + _require_env(TERMINAL_CUSTOM_BASE_URL_ENV), + TERMINAL_CUSTOM_BASE_URL_ENV, + ) + + +@pytest.fixture(scope="session") +def terminals_sdk( + terminals_e2e_api_key: str, + terminals_e2e_base_url: str, + terminals_partner_affiliate_api_key: Optional[str], +) -> Sdk: + """Return SDK isolated for terminal endpoint E2E tests.""" + resolver_kwargs: dict = { + "default_api_key": terminals_e2e_api_key, + } + if terminals_partner_affiliate_api_key: + resolver_kwargs["partner_affiliate_api_key"] = ( + terminals_partner_affiliate_api_key + ) + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + sdk.get_client().url = terminals_e2e_base_url + return sdk + + +@pytest.fixture(scope="session") +def terminals_group_id( + terminals_sdk: Sdk, + terminals_terminal_id: str, +) -> str: + """Return terminal group id used by terminal endpoint E2E tests.""" + return _resolve_terminal_group_id( + terminals_sdk=terminals_sdk, + terminal_id=terminals_terminal_id, + label="terminal endpoint E2E group id", + ) + + +def pytest_collection_modifyitems( + config: pytest.Config, # noqa: ARG001 + items: list[pytest.Item], +) -> None: + """Skip example E2E tests when the required credentials are missing.""" + has_default_e2e = bool(os.getenv(DEFAULT_E2E_API_KEY_ENV, "").strip()) + has_terminal_e2e = _has_terminal_e2e_env() + + if has_default_e2e and has_terminal_e2e: + return + + default_skip = pytest.mark.skip( + reason=f"E2E tests require {DEFAULT_E2E_API_KEY_ENV} (not set)", + ) + terminal_skip = pytest.mark.skip( + reason=( + "Terminal endpoint E2E tests require API_KEY, " + "MSP_SDK_CUSTOM_BASE_URL, and E2E_CLOUD_POS_TERMINAL_ID " + "(not set)" + ), + ) + for item in items: + if item.nodeid.startswith(TERMINAL_E2E_NODE_PREFIXES): + if has_terminal_e2e: + continue + item.add_marker(terminal_skip) + continue + + if not has_default_e2e: + item.add_marker(default_skip) diff --git a/tests/multisafepay/e2e/examples/terminal_group_manager/test_get_terminals_by_group.py b/tests/multisafepay/e2e/examples/terminal_group_manager/test_get_terminals_by_group.py new file mode 100644 index 0000000..6aaa295 --- /dev/null +++ b/tests/multisafepay/e2e/examples/terminal_group_manager/test_get_terminals_by_group.py @@ -0,0 +1,44 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""E2E coverage for examples/terminal_group_manager/get_terminals_by_group.py.""" + +import pytest + +from multisafepay.api.base.listings.listing_pager import ListingPager +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminal_groups.terminal_group_manager import ( + TerminalGroupManager, +) +from multisafepay.sdk import Sdk + + +@pytest.fixture(scope="module") +def terminal_group_manager(terminals_sdk: Sdk) -> TerminalGroupManager: + """Fixture that provides a TerminalGroupManager instance for testing.""" + return terminals_sdk.get_terminal_group_manager() + + +def test_get_terminals_by_group( + terminal_group_manager: TerminalGroupManager, + terminals_group_id: str, +) -> None: + """List terminals for a specific group using the example template flow.""" + response = terminal_group_manager.get_terminals_by_group( + terminal_group_id=terminals_group_id, + options={ + "limit": 10, + "page": 1, + }, + ) + + assert isinstance(response, CustomApiResponse) + assert response.get_status_code() == 200 + assert response.get_body_success() is True + assert isinstance(response.get_data(), ListingPager) diff --git a/tests/multisafepay/e2e/examples/terminal_manager/test_get_terminals.py b/tests/multisafepay/e2e/examples/terminal_manager/test_get_terminals.py new file mode 100644 index 0000000..b0b4ffd --- /dev/null +++ b/tests/multisafepay/e2e/examples/terminal_manager/test_get_terminals.py @@ -0,0 +1,38 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""E2E coverage for examples/terminal_manager/get_terminals.py.""" + +import pytest + +from multisafepay.api.base.listings.listing_pager import ListingPager +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminals.terminal_manager import TerminalManager +from multisafepay.sdk import Sdk + + +@pytest.fixture(scope="module") +def terminal_manager(terminals_sdk: Sdk) -> TerminalManager: + """Fixture that provides a TerminalManager instance for testing.""" + return terminals_sdk.get_terminal_manager() + + +def test_get_terminals(terminal_manager: TerminalManager) -> None: + """List terminals using the same flow as the terminal manager example.""" + response = terminal_manager.get_terminals( + options={ + "limit": 10, + "page": 1, + }, + ) + + assert isinstance(response, CustomApiResponse) + assert response.get_status_code() == 200 + assert response.get_body_success() is True + assert isinstance(response.get_data(), ListingPager) diff --git a/tests/multisafepay/unit/api/path/terminal_groups/__init__.py b/tests/multisafepay/unit/api/path/terminal_groups/__init__.py new file mode 100644 index 0000000..3f34622 --- /dev/null +++ b/tests/multisafepay/unit/api/path/terminal_groups/__init__.py @@ -0,0 +1,8 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Unit tests for terminal group path package.""" diff --git a/tests/multisafepay/unit/api/path/terminal_groups/test_unit_terminal_group_manager.py b/tests/multisafepay/unit/api/path/terminal_groups/test_unit_terminal_group_manager.py new file mode 100644 index 0000000..7e935b4 --- /dev/null +++ b/tests/multisafepay/unit/api/path/terminal_groups/test_unit_terminal_group_manager.py @@ -0,0 +1,107 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Unit tests for TerminalGroupManager.get_terminals_by_group behavior.""" + +from unittest.mock import MagicMock + +from multisafepay.api.base.response.api_response import ApiResponse +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminal_groups.terminal_group_manager import ( + TerminalGroupManager, +) +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope + +TERMINAL_GROUP_ID = "42" + + +def _build_listing_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": [ + { + "terminal_id": "T-001", + "name": "POS Terminal 1", + "group_id": 42, + "active": True, + "status": "active", + "provider": "CTAP", + }, + ], + "pager": { + "total": 1, + "offset": 0, + "limit": 10, + }, + }, + ) + + +def test_get_terminals_by_group_uses_partner_affiliate_scope() -> None: + """Send partner_affiliate auth scope for terminal group listing.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalGroupManager(client) + response = manager.get_terminals_by_group( + terminal_group_id=TERMINAL_GROUP_ID, + ) + + called_auth_scope = client.create_get_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_PARTNER_AFFILIATE, + ) + + +def test_get_terminals_by_group_encodes_group_id_in_endpoint() -> None: + """Verify terminal_group_id is included in the URL path.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalGroupManager(client) + manager.get_terminals_by_group(terminal_group_id=TERMINAL_GROUP_ID) + + called_endpoint = client.create_get_request.call_args.kwargs["endpoint"] + assert TERMINAL_GROUP_ID in called_endpoint + assert "json/terminal-groups/" in called_endpoint + + +def test_get_terminals_by_group_filters_options() -> None: + """Only pass allowed options (page, limit) to the API.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalGroupManager(client) + manager.get_terminals_by_group( + terminal_group_id=TERMINAL_GROUP_ID, + options={"page": 2, "limit": 5, "foo": "bar"}, + ) + + called_params = client.create_get_request.call_args.kwargs["params"] + assert called_params == {"page": 2, "limit": 5} + + +def test_get_terminals_by_group_defaults_empty_options() -> None: + """Use empty options dict when no options are provided.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalGroupManager(client) + manager.get_terminals_by_group(terminal_group_id=TERMINAL_GROUP_ID) + + called_params = client.create_get_request.call_args.kwargs["params"] + assert called_params == {} diff --git a/tests/multisafepay/unit/api/path/terminals/request/test_unit_create_terminal_request.py b/tests/multisafepay/unit/api/path/terminals/request/test_unit_create_terminal_request.py new file mode 100644 index 0000000..3ae2196 --- /dev/null +++ b/tests/multisafepay/unit/api/path/terminals/request/test_unit_create_terminal_request.py @@ -0,0 +1,63 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Unit tests for the terminal create request model.""" + +import pytest + +from multisafepay.api.paths.terminals.request.create_terminal_request import ( + CTAP_PROVIDER, + CreateTerminalRequest, +) +from multisafepay.exception.invalid_argument import InvalidArgumentException + + +def test_initializes_with_default_values() -> None: + """Initialize request model with empty/default values.""" + request = CreateTerminalRequest() + + assert request.provider is None + assert request.group_id is None + assert request.name is None + + +def test_add_provider_updates_value() -> None: + """Store a valid provider and return the current request object.""" + request = CreateTerminalRequest() + + returned = request.add_provider(CTAP_PROVIDER) + + assert request.provider == CTAP_PROVIDER + assert returned is request + + +def test_add_provider_raises_for_invalid_provider() -> None: + """Reject provider values that are not whitelisted.""" + request = CreateTerminalRequest() + + with pytest.raises(InvalidArgumentException, match="not a known provider"): + request.add_provider("UNKNOWN") + + +def test_add_group_id_updates_value() -> None: + """Store terminal group id and return current request object.""" + request = CreateTerminalRequest() + + returned = request.add_group_id("1234") + + assert request.group_id == "1234" + assert returned is request + + +def test_add_name_updates_value() -> None: + """Store terminal display name and return current request object.""" + request = CreateTerminalRequest() + + returned = request.add_name("Demo POS Terminal") + + assert request.name == "Demo POS Terminal" + assert returned is request diff --git a/tests/multisafepay/unit/api/path/terminals/response/test_unit_terminal_response.py b/tests/multisafepay/unit/api/path/terminals/response/test_unit_terminal_response.py new file mode 100644 index 0000000..536164a --- /dev/null +++ b/tests/multisafepay/unit/api/path/terminals/response/test_unit_terminal_response.py @@ -0,0 +1,94 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Unit tests for the terminal response model.""" + +from multisafepay.api.paths.terminals.response.terminal import Terminal + +TERMINAL_DATA = { + "id": "term-001", + "provider": "CTAP", + "name": "My Terminal", + "code": "T001", + "created": "2024-01-01T00:00:00", + "last_updated": "2024-06-01T00:00:00", + "manufacturer_id": "MFR-123", + "serial_number": "SN-456", + "active": True, + "group_id": 12345, + "country": "NL", +} + +EMPTY_TERMINAL_DATA = {field: None for field in TERMINAL_DATA} + + +def _assert_terminal_data(terminal: Terminal, expected: dict) -> None: + """Assert terminal attributes against expected fixture data.""" + for field, expected_value in expected.items(): + assert getattr(terminal, field) == expected_value + + +def test_initializes_with_all_fields(): + """ + Test that the Terminal object initializes correctly with all fields. + + This test verifies that the Terminal object stores the correct values for + all its attributes when instantiated with explicit data. + """ + terminal = Terminal(**TERMINAL_DATA) + + _assert_terminal_data(terminal, TERMINAL_DATA) + + +def test_initializes_with_none_values(): + """ + Test that the Terminal object initializes correctly with None values. + + This test verifies that all attributes default to None when the Terminal + object is instantiated without any arguments. + """ + terminal = Terminal() + + _assert_terminal_data(terminal, EMPTY_TERMINAL_DATA) + + +def test_from_dict_creates_instance_from_dict(): + """ + Test that the from_dict method creates a Terminal from a valid dictionary. + + This test verifies that from_dict correctly maps all dictionary keys to + the corresponding Terminal attributes. + """ + terminal: Terminal | None = Terminal.from_dict(TERMINAL_DATA) + + assert terminal is not None + _assert_terminal_data(terminal, TERMINAL_DATA) + + +def test_from_dict_returns_none_for_none_input(): + """ + Test that the from_dict method returns None when the input is None. + + This test verifies that from_dict returns None when None is provided + as the input dictionary. + """ + terminal = Terminal.from_dict(None) + assert terminal is None + + +def test_from_dict_handles_missing_fields(): + """ + Test that the from_dict method handles missing fields by setting them to None. + + This test verifies that from_dict correctly creates a Terminal from a + dictionary with missing fields, resulting in None values for those attributes. + """ + data = {} + terminal: Terminal | None = Terminal.from_dict(data) + + assert terminal is not None + _assert_terminal_data(terminal, EMPTY_TERMINAL_DATA) diff --git a/tests/multisafepay/unit/api/path/terminals/test_unit_terminal_manager.py b/tests/multisafepay/unit/api/path/terminals/test_unit_terminal_manager.py new file mode 100644 index 0000000..3656878 --- /dev/null +++ b/tests/multisafepay/unit/api/path/terminals/test_unit_terminal_manager.py @@ -0,0 +1,150 @@ +# Copyright (c) MultiSafepay, Inc. All rights reserved. + +# This file is licensed under the Open Software License (OSL) version 3.0. +# For a copy of the license, see the LICENSE.txt file in the project root. + +# See the DISCLAIMER.md file for disclaimer details. + +"""Unit tests for TerminalManager.create_terminal and get_terminals behavior.""" + +from unittest.mock import MagicMock + +from multisafepay.api.base.response.api_response import ApiResponse +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.terminals.request.create_terminal_request import ( + CreateTerminalRequest, +) +from multisafepay.api.paths.terminals.response.terminal import Terminal +from multisafepay.api.paths.terminals.terminal_manager import TerminalManager +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope + + +def _build_terminal_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "terminal_id": "T-001", + "name": "Demo POS Terminal", + "group_id": 42, + "active": True, + "status": "active", + "provider": "CTAP", + }, + }, + ) + + +def _build_listing_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": [ + { + "terminal_id": "T-001", + "name": "Terminal 1", + "group_id": 42, + "active": True, + "status": "active", + "provider": "CTAP", + }, + ], + "pager": { + "total": 1, + "offset": 0, + "limit": 10, + }, + }, + ) + + +def test_create_terminal_sends_post_to_correct_endpoint() -> None: + """create_terminal posts to json/terminals with no auth scope.""" + client = MagicMock() + client.create_post_request.return_value = _build_terminal_api_response() + + request = ( + CreateTerminalRequest() + .add_provider("CTAP") + .add_group_id(42) + .add_name("Demo POS Terminal") + ) + + manager = TerminalManager(client) + response = manager.create_terminal(request) + + called_endpoint = client.create_post_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Terminal) + assert response.get_data().terminal_id == "T-001" + assert called_endpoint == "json/terminals" + + +def test_create_terminal_serializes_request_body() -> None: + """Verify the request body is serialized as JSON.""" + client = MagicMock() + client.create_post_request.return_value = _build_terminal_api_response() + + request = ( + CreateTerminalRequest() + .add_provider("CTAP") + .add_group_id(42) + .add_name("Demo POS Terminal") + ) + + manager = TerminalManager(client) + manager.create_terminal(request) + + called_body = client.create_post_request.call_args.kwargs["request_body"] + assert '"provider": "CTAP"' in called_body + assert '"group_id": 42' in called_body + + +def test_get_terminals_uses_partner_affiliate_scope() -> None: + """get_terminals sends partner_affiliate auth scope.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalManager(client) + response = manager.get_terminals(options={"page": 1, "limit": 10}) + + called_auth_scope = client.create_get_request.call_args.kwargs.get( + "auth_scope", + ) + + assert isinstance(response, CustomApiResponse) + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_PARTNER_AFFILIATE, + ) + + +def test_get_terminals_filters_options() -> None: + """Only pass allowed options (page, limit) to the API.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalManager(client) + manager.get_terminals(options={"page": 1, "limit": 5, "invalid": "x"}) + + called_params = client.create_get_request.call_args.args[1] + assert called_params == {"page": 1, "limit": 5} + + +def test_get_terminals_defaults_empty_options() -> None: + """Use empty options dict when no options are provided.""" + client = MagicMock() + client.create_get_request.return_value = _build_listing_api_response() + + manager = TerminalManager(client) + manager.get_terminals() + + called_params = client.create_get_request.call_args.args[1] + assert called_params == {} diff --git a/tests/multisafepay/unit/test_unit_sdk.py b/tests/multisafepay/unit/test_unit_sdk.py index f160905..b667891 100644 --- a/tests/multisafepay/unit/test_unit_sdk.py +++ b/tests/multisafepay/unit/test_unit_sdk.py @@ -7,9 +7,15 @@ """Unit tests for SDK-level environment/base URL guardrails.""" +from unittest.mock import MagicMock + import pytest from multisafepay import Sdk +from multisafepay.api.paths.terminal_groups.terminal_group_manager import ( + TerminalGroupManager, +) +from multisafepay.api.paths.terminals.terminal_manager import TerminalManager from multisafepay.client.client import Client from multisafepay.client.credential_resolver import ScopedCredentialResolver @@ -137,3 +143,25 @@ def test_sdk_uses_credential_resolver_with_custom_transport() -> None: assert sdk.get_client().transport is transport assert transport.headers["Authorization"] == f"Bearer {DEFAULT_API_KEY}" + + +def test_sdk_returns_terminal_manager() -> None: + """Expose TerminalManager through SDK convenience getter.""" + sdk = Sdk( + api_key="mock_api_key", + is_production=False, + transport=MagicMock(), + ) + + assert isinstance(sdk.get_terminal_manager(), TerminalManager) + + +def test_sdk_returns_terminal_group_manager() -> None: + """Expose TerminalGroupManager through SDK convenience getter.""" + sdk = Sdk( + api_key="mock_api_key", + is_production=False, + transport=MagicMock(), + ) + + assert isinstance(sdk.get_terminal_group_manager(), TerminalGroupManager)