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/client/__init__.py b/src/multisafepay/client/__init__.py index a14961a..922e1d2 100644 --- a/src/multisafepay/client/__init__.py +++ b/src/multisafepay/client/__init__.py @@ -2,8 +2,10 @@ from multisafepay.client.api_key import ApiKey from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import ScopedCredentialResolver __all__ = [ "ApiKey", "Client", + "ScopedCredentialResolver", ] diff --git a/src/multisafepay/client/client.py b/src/multisafepay/client/client.py index d6d5e00..2c9a3dc 100644 --- a/src/multisafepay/client/client.py +++ b/src/multisafepay/client/client.py @@ -16,6 +16,11 @@ from ..exception.api import ApiException from .api_key import ApiKey +from .credential_resolver import ( + AuthScope, + CredentialResolver, + ScopedCredentialResolver, +) class Client: @@ -39,6 +44,14 @@ class Client: CUSTOM_BASE_URL_ENV = "MSP_SDK_CUSTOM_BASE_URL" ALLOW_CUSTOM_BASE_URL_ENV = "MSP_SDK_ALLOW_CUSTOM_BASE_URL" + AUTH_SCOPE_DEFAULT = ScopedCredentialResolver.AUTH_SCOPE_DEFAULT + AUTH_SCOPE_PARTNER_AFFILIATE = ( + ScopedCredentialResolver.AUTH_SCOPE_PARTNER_AFFILIATE + ) + AUTH_SCOPE_TERMINAL_GROUP = ( + ScopedCredentialResolver.AUTH_SCOPE_TERMINAL_GROUP + ) + METHOD_POST = "POST" METHOD_GET = "GET" METHOD_PATCH = "PATCH" @@ -46,18 +59,20 @@ class Client: def __init__( self: "Client", - api_key: str, - is_production: bool, + api_key: Optional[str] = None, + is_production: bool = False, transport: Optional[HTTPTransport] = None, locale: str = "en_US", base_url: Optional[str] = None, + credential_resolver: Optional[CredentialResolver] = None, ) -> None: """ Initialize the Client. Parameters ---------- - api_key (str): The API key for authentication. + api_key (Optional[str]): The API key for authentication. + Optional only when `credential_resolver` is provided. is_production (bool): Flag indicating if the client is in production mode. transport (Optional[HTTPTransport], optional): Custom HTTP transport implementation. Defaults to RequestsTransport if not provided. @@ -65,9 +80,17 @@ def __init__( base_url (Optional[str], optional): Custom API base URL. Only allowed when running with `MSP_SDK_BUILD_PROFILE=dev` and `MSP_SDK_ALLOW_CUSTOM_BASE_URL=1`. + credential_resolver (Optional[CredentialResolver], optional): + Resolver used to derive API keys by auth scope. """ - self.api_key = ApiKey(api_key=api_key) + if api_key is None and credential_resolver is None: + raise ValueError( + "api_key is required when credential_resolver is not provided.", + ) + + self.api_key = ApiKey(api_key=api_key) if api_key is not None else None + self.credential_resolver = credential_resolver self.url = self._resolve_base_url( is_production=is_production, explicit_base_url=base_url, @@ -123,6 +146,7 @@ def create_get_request( endpoint: str, params: dict[str, Any] = None, context: Optional[dict[str, Any]] = None, + auth_scope: Optional[AuthScope] = None, ) -> ApiResponse: """ Create a GET request. @@ -143,6 +167,7 @@ def create_get_request( self.METHOD_GET, url, context=context, + auth_scope=auth_scope, ) def create_post_request( @@ -151,6 +176,7 @@ def create_post_request( params: dict[str, Any] = None, request_body: str = None, context: Optional[dict[str, Any]] = None, + auth_scope: Optional[AuthScope] = None, ) -> ApiResponse: """ Create a POST request. @@ -173,6 +199,7 @@ def create_post_request( url, request_body=request_body, context=context, + auth_scope=auth_scope, ) def create_patch_request( @@ -181,6 +208,7 @@ def create_patch_request( params: dict[str, Any] = None, request_body: str = None, context: Optional[dict[str, Any]] = None, + auth_scope: Optional[AuthScope] = None, ) -> ApiResponse: """ Create a PATCH request. @@ -203,6 +231,7 @@ def create_patch_request( url, request_body=request_body, context=context, + auth_scope=auth_scope, ) def create_delete_request( @@ -210,6 +239,7 @@ def create_delete_request( endpoint: str, params: dict[str, Any] = None, context: Optional[dict[str, Any]] = None, + auth_scope: Optional[AuthScope] = None, ) -> ApiResponse: """ Create a DELETE request. @@ -226,7 +256,12 @@ def create_delete_request( """ url = self._build_url(endpoint, params) - return self._create_request(self.METHOD_DELETE, url, context=context) + return self._create_request( + self.METHOD_DELETE, + url, + context=context, + auth_scope=auth_scope, + ) def _build_url( self: "Client", @@ -255,12 +290,33 @@ def _build_url( ) return f"{self.url}{endpoint}?{query_string}" + def _resolve_api_key( + self: "Client", + auth_scope: Optional[AuthScope], + ) -> str: + if self.credential_resolver is not None: + resolved_scope = auth_scope or AuthScope( + scope=self.AUTH_SCOPE_DEFAULT, + ) + return self.credential_resolver.resolve( + auth_scope=resolved_scope.scope, + group_id=resolved_scope.group_id, + ) + + if self.api_key is None: + raise ValueError( + "api_key is required when credential_resolver is not provided.", + ) + + return self.api_key.get() + def _create_request( self: "Client", method: str, url: str, request_body: Optional[dict[str, Any]] = None, context: Optional[dict[str, Any]] = None, + auth_scope: Optional[AuthScope] = None, ) -> ApiResponse: """ Create and send an HTTP request. @@ -277,8 +333,9 @@ def _create_request( ApiResponse: The API response. """ + api_key = self._resolve_api_key(auth_scope) headers = { - "Authorization": "Bearer " + self.api_key.get(), + "Authorization": "Bearer " + api_key, "accept-encoding": "application/json", "Content-Type": "application/json", } diff --git a/src/multisafepay/client/credential_resolver.py b/src/multisafepay/client/credential_resolver.py new file mode 100644 index 0000000..860bcaa --- /dev/null +++ b/src/multisafepay/client/credential_resolver.py @@ -0,0 +1,106 @@ +# 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. + +"""Credential resolver contracts and default scoped resolver.""" + +from dataclasses import dataclass +from typing import Optional, Protocol + + +@dataclass(frozen=True) +class AuthScope: + """Auth scope selection payload for credential resolution.""" + + scope: str + group_id: Optional[str] = None + + +class CredentialResolver(Protocol): + """Protocol for resolving API keys by auth scope and context.""" + + def resolve( + self: "CredentialResolver", + auth_scope: str, + group_id: Optional[str] = None, + ) -> str: + """Resolve the API key to use for a given scope and context.""" + + +class ScopedCredentialResolver: + """Default resolver implementation for account, partner and group scopes.""" + + AUTH_SCOPE_DEFAULT = "default_account" + AUTH_SCOPE_PARTNER_AFFILIATE = "partner_affiliate" + AUTH_SCOPE_TERMINAL_GROUP = "terminal_group" + + def __init__( + self: "ScopedCredentialResolver", + default_api_key: str, + partner_affiliate_api_key: Optional[str] = None, + terminal_group_api_keys: Optional[dict[str, str]] = None, + ) -> None: + """ + Initialize a scoped credential resolver. + + Parameters + ---------- + default_api_key (str): Fallback/default account API key. + partner_affiliate_api_key (Optional[str]): Partner/affiliate API key. + terminal_group_api_keys (Optional[dict[str, str]]): Mapping of + terminal_group_id to API key. + + """ + self.default_api_key = (default_api_key or "").strip() + self.partner_affiliate_api_key = ( + partner_affiliate_api_key or "" + ).strip() or None + self.terminal_group_api_keys = { + group_id: api_key.strip() + for group_id, api_key in (terminal_group_api_keys or {}).items() + if api_key and api_key.strip() + } + + if ( + not self.default_api_key + and self.partner_affiliate_api_key is None + and not self.terminal_group_api_keys + ): + raise ValueError( + "ScopedCredentialResolver requires at least one API key.", + ) + + def resolve( + self: "ScopedCredentialResolver", + auth_scope: str, + group_id: Optional[str] = None, + ) -> str: + """Resolve API key for the given scope and auth context.""" + if auth_scope == self.AUTH_SCOPE_TERMINAL_GROUP: + if not group_id: + raise ValueError( + "Missing terminal_group_id in auth scope.", + ) + api_key = self.terminal_group_api_keys.get(group_id) + if not api_key: + raise ValueError( + "No API key configured for terminal_group_id " + f"'{group_id}'.", + ) + return api_key + + if auth_scope == self.AUTH_SCOPE_PARTNER_AFFILIATE: + api_key = self.partner_affiliate_api_key or self.default_api_key + if not api_key: + raise ValueError( + "No API key configured for partner_affiliate scope.", + ) + return api_key + + if not self.default_api_key: + raise ValueError("No API key configured for default scope.") + + return self.default_api_key diff --git a/src/multisafepay/sdk.py b/src/multisafepay/sdk.py index b22dc3b..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, ) @@ -26,6 +30,7 @@ from .api.paths.me.me_manager import MeManager from .api.paths.recurring.recurring_manager import RecurringManager from .client.client import Client +from .client.credential_resolver import CredentialResolver class Sdk: @@ -38,19 +43,21 @@ class Sdk: def __init__( self: "Sdk", - api_key: str, - is_production: bool, + api_key: Optional[str] = None, + is_production: bool = False, transport: Optional[HTTPTransport] = None, locale: str = "en_US", base_url: Optional[str] = None, + credential_resolver: Optional[CredentialResolver] = None, ) -> None: """ Initialize the SDK with the provided configuration. Parameters ---------- - api_key : str + api_key : Optional[str] The API key for authenticating with the MultiSafePay API. + Optional only when `credential_resolver` is provided. is_production : bool Flag indicating whether to use the production environment. transport : Optional[HTTPTransport], optional @@ -60,14 +67,17 @@ def __init__( The locale to use for requests, by default "en_US". base_url : Optional[str], optional Custom API base URL (dev-only guardrails apply), by default None. + credential_resolver : Optional[CredentialResolver], optional + Strategy for resolving API keys per auth scope, by default None. """ self.client = Client( - api_key.strip(), - is_production, - transport, - locale, - base_url, + api_key=api_key, + is_production=is_production, + transport=transport, + locale=locale, + base_url=base_url, + credential_resolver=credential_resolver, ) self.recurring_manager = RecurringManager(self.client) @@ -191,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/client/test_unit_client.py b/tests/multisafepay/unit/client/test_unit_client.py index 92af510..b4bbc90 100644 --- a/tests/multisafepay/unit/client/test_unit_client.py +++ b/tests/multisafepay/unit/client/test_unit_client.py @@ -11,10 +11,61 @@ import pytest from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import ( + ScopedCredentialResolver, +) from multisafepay.transport import RequestsTransport requests = pytest.importorskip("requests") +DEFAULT_API_KEY = "default_api_key" +TERMINAL_GROUP_ID = "Default" +TERMINAL_GROUP_API_KEY = "terminal_group_api_key" +ORDERS_ENDPOINT = "json/orders" +API_KEY_REQUIRED_ERROR = "api_key is required" + + +class _FakeResponse: + """Small HTTP response stub for unit tests.""" + + status_code = 200 + headers = {} + + @staticmethod + def json() -> dict: + return { + "success": True, + "data": {}, + } + + @staticmethod + def raise_for_status() -> None: + return + + +class _CaptureTransport: + """Transport stub that captures the request headers.""" + + def __init__(self: "_CaptureTransport") -> None: + self.headers = {} + + def request(self: "_CaptureTransport", **kwargs: dict) -> _FakeResponse: + self.headers = kwargs.get("headers", {}) + return _FakeResponse() + + +def _build_resolver_client( + resolver: ScopedCredentialResolver, + transport: _CaptureTransport, +) -> Client: + """Build a client configured for resolver-based auth tests.""" + return Client( + api_key=None, + is_production=False, + transport=transport, + credential_resolver=resolver, + ) + def test_initializes_with_default_requests_transport(): """Test that the Client initializes with the default requests transport.""" @@ -211,3 +262,71 @@ def test_rejects_custom_base_url_without_netloc( is_production=False, base_url="https:///v1", ) + + +def test_create_get_request_sends_authorization_header() -> None: + """GET request includes Bearer authorization header.""" + transport = _CaptureTransport() + client = Client( + api_key="test_key", + is_production=False, + transport=transport, + ) + client.create_get_request("json/orders") + assert transport.headers["Authorization"] == "Bearer test_key" + + +def test_create_post_request_sends_authorization_header() -> None: + """POST request includes Bearer authorization header.""" + transport = _CaptureTransport() + client = Client( + api_key="test_key", + is_production=False, + transport=transport, + ) + client.create_post_request("json/orders", request_body='{"foo":"bar"}') + assert transport.headers["Authorization"] == "Bearer test_key" + + +def test_create_patch_request_sends_authorization_header() -> None: + """PATCH request includes Bearer authorization header.""" + transport = _CaptureTransport() + client = Client( + api_key="test_key", + is_production=False, + transport=transport, + ) + client.create_patch_request("json/orders/1", request_body='{"foo":"bar"}') + assert transport.headers["Authorization"] == "Bearer test_key" + + +def test_create_delete_request_sends_authorization_header() -> None: + """DELETE request includes Bearer authorization header.""" + transport = _CaptureTransport() + client = Client( + api_key="test_key", + is_production=False, + transport=transport, + ) + client.create_delete_request("json/recurring/1") + assert transport.headers["Authorization"] == "Bearer test_key" + + +def test_resolve_api_key_uses_credential_resolver() -> None: + """Prefer credential resolver when both api_key and resolver exist.""" + transport = _CaptureTransport() + resolver = ScopedCredentialResolver(default_api_key="resolver_key") + client = _build_resolver_client(resolver, transport) + client.create_get_request("json/orders") + assert transport.headers["Authorization"] == "Bearer resolver_key" + + +def test_resolve_api_key_raises_without_key_or_resolver() -> None: + """Raise ValueError when no api_key or resolver is configured.""" + with pytest.raises(ValueError, match=API_KEY_REQUIRED_ERROR): + Client( + api_key=None, + is_production=False, + transport=_CaptureTransport(), + credential_resolver=None, + ) diff --git a/tests/multisafepay/unit/client/test_unit_credential_resolver.py b/tests/multisafepay/unit/client/test_unit_credential_resolver.py new file mode 100644 index 0000000..52d7f0e --- /dev/null +++ b/tests/multisafepay/unit/client/test_unit_credential_resolver.py @@ -0,0 +1,134 @@ +# 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 scoped credential resolver behavior.""" + +import pytest + +from multisafepay.client.credential_resolver import ScopedCredentialResolver + +DEFAULT_API_KEY = "default_api_key" +PARTNER_API_KEY = "partner_api_key" +TERMINAL_GROUP_ID = "Default" +TERMINAL_GROUP_API_KEY = "terminal_group_api_key" +MISSING_GROUP_ID_ERROR = "Missing terminal_group_id" +NO_DEFAULT_SCOPE_ERROR = "No API key configured for default scope" + + +def _resolver_with_terminal_group() -> ScopedCredentialResolver: + """Create resolver fixture data for terminal-group scope tests.""" + return ScopedCredentialResolver( + default_api_key=DEFAULT_API_KEY, + terminal_group_api_keys={ + TERMINAL_GROUP_ID: TERMINAL_GROUP_API_KEY, + }, + ) + + +def test_rejects_resolver_without_any_api_key() -> None: + """Require at least one API key across all resolver sources.""" + with pytest.raises( + ValueError, + match="requires at least one API key", + ): + ScopedCredentialResolver(default_api_key="") + + +def test_resolves_default_scope_with_default_api_key() -> None: + """Resolve default scope using the configured default API key.""" + resolver = ScopedCredentialResolver(default_api_key=DEFAULT_API_KEY) + + assert ( + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_DEFAULT) + == DEFAULT_API_KEY + ) + + +def test_resolves_partner_scope_with_partner_api_key() -> None: + """Prefer partner key for partner_affiliate scope.""" + resolver = ScopedCredentialResolver( + default_api_key=DEFAULT_API_KEY, + partner_affiliate_api_key=PARTNER_API_KEY, + ) + + assert ( + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_PARTNER_AFFILIATE) + == PARTNER_API_KEY + ) + + +def test_resolves_terminal_group_scope_with_group_key() -> None: + """Resolve terminal_group scope using group-specific API key mapping.""" + resolver = _resolver_with_terminal_group() + + assert ( + resolver.resolve( + ScopedCredentialResolver.AUTH_SCOPE_TERMINAL_GROUP, + group_id=TERMINAL_GROUP_ID, + ) + == TERMINAL_GROUP_API_KEY + ) + + +def test_raises_for_terminal_group_scope_without_group_id() -> None: + """Reject terminal_group scope when group_id is missing.""" + resolver = _resolver_with_terminal_group() + + with pytest.raises(ValueError, match=MISSING_GROUP_ID_ERROR): + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_TERMINAL_GROUP) + + +def test_raises_for_default_scope_without_default_key() -> None: + """Reject default scope when no default key is configured.""" + resolver = ScopedCredentialResolver( + default_api_key="", + partner_affiliate_api_key=PARTNER_API_KEY, + ) + + with pytest.raises( + ValueError, + match=NO_DEFAULT_SCOPE_ERROR, + ): + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_DEFAULT) + + +def test_resolves_partner_scope_falls_back_to_default_key() -> None: + """Fall back to default_api_key for partner scope when no partner key.""" + resolver = ScopedCredentialResolver(default_api_key=DEFAULT_API_KEY) + + assert ( + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_PARTNER_AFFILIATE) + == DEFAULT_API_KEY + ) + + +def test_raises_for_unknown_terminal_group_id() -> None: + """Reject terminal_group scope when the group_id is not configured.""" + resolver = _resolver_with_terminal_group() + + with pytest.raises(ValueError, match="No API key configured"): + resolver.resolve( + ScopedCredentialResolver.AUTH_SCOPE_TERMINAL_GROUP, + group_id="unknown_group", + ) + + +def test_strips_whitespace_from_api_keys() -> None: + """Strip leading/trailing whitespace from provided API keys.""" + resolver = ScopedCredentialResolver( + default_api_key=" key_with_spaces ", + partner_affiliate_api_key=" partner_spaces ", + ) + + assert ( + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_DEFAULT) + == "key_with_spaces" + ) + assert ( + resolver.resolve(ScopedCredentialResolver.AUTH_SCOPE_PARTNER_AFFILIATE) + == "partner_spaces" + ) diff --git a/tests/multisafepay/unit/test_unit_sdk.py b/tests/multisafepay/unit/test_unit_sdk.py index 038338e..b667891 100644 --- a/tests/multisafepay/unit/test_unit_sdk.py +++ b/tests/multisafepay/unit/test_unit_sdk.py @@ -7,10 +7,48 @@ """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 + +DEFAULT_API_KEY = "resolver_api_key" + + +class _FakeResponse: + """Small HTTP response stub for SDK transport tests.""" + + status_code = 200 + headers = {} + + @staticmethod + def json() -> dict: + return { + "success": True, + "data": {}, + } + + @staticmethod + def raise_for_status() -> None: + return + + +class _CaptureTransport: + """Transport stub that captures outbound request headers.""" + + def __init__(self: "_CaptureTransport") -> None: + self.headers = {} + + def request(self: "_CaptureTransport", **kwargs: dict) -> _FakeResponse: + self.headers = kwargs.get("headers", {}) + return _FakeResponse() def test_sdk_uses_test_url_by_default(monkeypatch: pytest.MonkeyPatch): @@ -64,3 +102,66 @@ def test_sdk_blocks_custom_base_url_in_release( is_production=False, base_url="https://dev-api.multisafepay.test/v1", ) + + +def test_sdk_allows_resolver_only_initialization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Allow constructing SDK without api_key when resolver is provided.""" + monkeypatch.delenv("MSP_SDK_BUILD_PROFILE", raising=False) + monkeypatch.delenv("MSP_SDK_CUSTOM_BASE_URL", raising=False) + monkeypatch.delenv("MSP_SDK_ALLOW_CUSTOM_BASE_URL", raising=False) + + resolver = ScopedCredentialResolver(default_api_key="resolver_api_key") + + sdk = Sdk( + is_production=False, + credential_resolver=resolver, + ) + + assert sdk.get_client().url == Client.TEST_URL + + +def test_sdk_requires_api_key_or_resolver() -> None: + """Reject SDK initialization when both api_key and resolver are missing.""" + with pytest.raises(ValueError, match="api_key is required"): + Sdk(is_production=False) + + +def test_sdk_uses_credential_resolver_with_custom_transport() -> None: + """Wire resolver + transport together and use resolved auth header.""" + transport = _CaptureTransport() + resolver = ScopedCredentialResolver(default_api_key=DEFAULT_API_KEY) + + sdk = Sdk( + is_production=False, + transport=transport, + credential_resolver=resolver, + ) + + sdk.get_client().create_get_request("json/orders") + + 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)