diff --git a/examples/order_manager/cancel.py b/examples/order_manager/cancel.py new file mode 100644 index 0000000..6b7091b --- /dev/null +++ b/examples/order_manager/cancel.py @@ -0,0 +1,109 @@ +# 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. + +"""Create a Cloud POS order, wait 5 seconds, and cancel it.""" + +import os +import time + +from dotenv import load_dotenv + +from multisafepay import Sdk +from multisafepay.api.paths.orders.request import OrderRequest +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + + +def _get_first_env(*names: str) -> str: + for name in names: + value = os.getenv(name, "").strip() + if value: + return value + + return "" + + +def _require_first_env(*names: str) -> str: + value = _get_first_env(*names) + if value: + return value + + raise RuntimeError( + f"Missing required environment variable. Set one of: {', '.join(names)}", + ) + + +DEFAULT_ACCOUNT_API_KEY = _require_first_env("API_KEY", "E2E_API_KEY") +TERMINAL_GROUP_DEFAULT_API_KEY = _require_first_env( + "TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", + "E2E_TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", +) +CLOUD_POS_TERMINAL_GROUP_ID = ( + os.getenv("CLOUD_POS_TERMINAL_GROUP_ID", "Default").strip() + or "Default" +) +TERMINAL_ID = _require_first_env( + "CLOUD_POS_TERMINAL_ID", + "E2E_CLOUD_POS_TERMINAL_ID", +) + +if __name__ == "__main__": + # This example executes Cloud POS calls with terminal-group scope. + scoped_terminal_group_id = CLOUD_POS_TERMINAL_GROUP_ID + resolver_kwargs = { + "default_api_key": DEFAULT_ACCOUNT_API_KEY, + } + if scoped_terminal_group_id: + resolver_kwargs["terminal_group_api_keys"] = { + scoped_terminal_group_id: TERMINAL_GROUP_DEFAULT_API_KEY, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + order_manager = multisafepay_sdk.get_order_manager() + + order_request = ( + OrderRequest() + .add_type("redirect") + .add_order_id(f"cloud-pos-cancel-{int(time.time())}") + .add_description("Cloud POS cancel order") + .add_amount(100) + .add_currency("EUR") + .add_gateway_info( + { + "terminal_id": TERMINAL_ID, + }, + ) + ) + + create_response = order_manager.create( + order_request, + terminal_group_id=scoped_terminal_group_id, + ) + order = create_response.get_data() + + if order is None or not order.order_id: + raise RuntimeError("Order creation did not return order_id") + + order_id = order.order_id + print(f"Created Cloud POS order: {order_id}") + print("Waiting 5 seconds before cancel...") + time.sleep(5) + + cancel_response = order_manager.cancel_transaction( + order_id, + terminal_group_id=scoped_terminal_group_id, + ) + + print(f"Canceled Cloud POS order: {order_id}") + print(cancel_response.get_data()) diff --git a/examples/order_manager/cloud_pos_order.py b/examples/order_manager/cloud_pos_order.py new file mode 100644 index 0000000..602f22c --- /dev/null +++ b/examples/order_manager/cloud_pos_order.py @@ -0,0 +1,102 @@ +# 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. + +"""Create a Cloud POS order using terminal-group scoped authentication.""" + +import os +import time + +from dotenv import load_dotenv +from multisafepay import Sdk +from multisafepay.api.paths.orders.request import OrderRequest +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + + +def _get_first_env(*names: str) -> str: + for name in names: + value = os.getenv(name, "").strip() + if value: + return value + + return "" + + +def _require_first_env(*names: str) -> str: + value = _get_first_env(*names) + if value: + return value + + raise RuntimeError( + f"Missing required environment variable. Set one of: {', '.join(names)}", + ) + + +DEFAULT_ACCOUNT_API_KEY = _require_first_env("API_KEY", "E2E_API_KEY") +TERMINAL_GROUP_DEFAULT_API_KEY = _require_first_env( + "TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", + "E2E_TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", +) +CLOUD_POS_TERMINAL_GROUP_ID = ( + os.getenv("CLOUD_POS_TERMINAL_GROUP_ID", "Default").strip() + or "Default" +) +TERMINAL_ID = _require_first_env( + "CLOUD_POS_TERMINAL_ID", + "E2E_CLOUD_POS_TERMINAL_ID", +) + +if __name__ == "__main__": + # This example executes Cloud POS calls with terminal-group scope. + scoped_terminal_group_id = CLOUD_POS_TERMINAL_GROUP_ID + resolver_kwargs = { + "default_api_key": DEFAULT_ACCOUNT_API_KEY, + } + if scoped_terminal_group_id: + resolver_kwargs["terminal_group_api_keys"] = { + scoped_terminal_group_id: TERMINAL_GROUP_DEFAULT_API_KEY, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + order_manager = multisafepay_sdk.get_order_manager() + + order_id = f"cloud-pos-{int(time.time())}" + + order_request = ( + OrderRequest() + .add_type("redirect") + .add_order_id(order_id) + .add_description("Cloud POS order") + .add_amount(100) + .add_currency("EUR") + .add_gateway_info( + { + "terminal_id": TERMINAL_ID, + }, + ) + ) + + create_response = order_manager.create( + order_request, + terminal_group_id=scoped_terminal_group_id, + ) + order = create_response.get_data() + + if order is None: + raise RuntimeError("Order creation did not return order data") + + print(f"Created Cloud POS order: {order.order_id}") + + if order.payment_url: + print(f"Payment URL: {order.payment_url}") diff --git a/examples/pos_manager/get_receipt.py b/examples/pos_manager/get_receipt.py new file mode 100644 index 0000000..cc87cc6 --- /dev/null +++ b/examples/pos_manager/get_receipt.py @@ -0,0 +1,181 @@ +# 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. + +"""Fetch the receipt for an existing Cloud POS order.""" + +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() + + +def _get_first_env(*names: str) -> str: + for name in names: + value = os.getenv(name, "").strip() + if value: + return value + + return "" + + +def _require_first_env(*names: str) -> str: + value = _get_first_env(*names) + if value: + return value + + raise RuntimeError( + f"Missing required environment variable. Set one of: {', '.join(names)}", + ) + + +DEFAULT_ACCOUNT_API_KEY = _require_first_env("API_KEY", "E2E_API_KEY") +TERMINAL_GROUP_DEFAULT_API_KEY = _require_first_env( + "TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", + "E2E_TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", +) +CLOUD_POS_TERMINAL_GROUP_ID = ( + os.getenv("CLOUD_POS_TERMINAL_GROUP_ID", "Default").strip() + or "Default" +) +ORDER_ID = _require_first_env("CLOUD_POS_ORDER_ID", "POS_ORDER_ID") + +if __name__ == "__main__": + credential_resolver = ScopedCredentialResolver( + default_api_key=DEFAULT_ACCOUNT_API_KEY, + terminal_group_api_keys={ + CLOUD_POS_TERMINAL_GROUP_ID: TERMINAL_GROUP_DEFAULT_API_KEY, + }, + ) + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + pos_manager = multisafepay_sdk.get_pos_manager() + + receipt_response = pos_manager.get_receipt( + order_id=ORDER_ID, + terminal_group_id=CLOUD_POS_TERMINAL_GROUP_ID, + ) + print(receipt_response.get_data())# 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 +import time + +from dotenv import load_dotenv + +from multisafepay import Sdk +from multisafepay.api.paths.orders.request import OrderRequest +from multisafepay.client import ScopedCredentialResolver + +# Load environment variables from a .env file +load_dotenv() + +default_account_api_key = (os.getenv("API_KEY") or "").strip() +terminal_group_default_api_key = (os.getenv("TERMINAL_GROUP_API_KEY_GROUP_DEFAULT") or "").strip() +partner_affiliate_api_key = (os.getenv("PARTNER_API_KEY") or "").strip() +terminal_group_id = os.getenv("CLOUD_POS_TERMINAL_GROUP_ID", "Default") +terminal_id = os.getenv("CLOUD_POS_TERMINAL_ID", "") + + +def _payload_get(payload: object, key: str) -> object: + """Read a payload field from either a dict or a response model.""" + if isinstance(payload, dict): + return payload.get(key) + return getattr(payload, key, None) + + +def _is_completed_event(event: object) -> bool: + """Return True when the SSE payload indicates a completed payment.""" + payload = getattr(event, "data", None) + status = _payload_get(payload, "status") + if isinstance(status, str) and status.lower() == "completed": + return True + + nested_payload = _payload_get(payload, "data") + nested_status = _payload_get(nested_payload, "status") + if ( + isinstance(nested_status, str) + and nested_status.lower() == "completed" + ): + return True + + return False + +if __name__ == "__main__": + # This example executes Cloud POS calls with terminal-group scope. + scoped_terminal_group_id = terminal_group_id + resolver_kwargs = { + "default_api_key": default_account_api_key, + "partner_affiliate_api_key": partner_affiliate_api_key, + } + if scoped_terminal_group_id: + resolver_kwargs["terminal_group_api_keys"] = { + scoped_terminal_group_id: terminal_group_default_api_key, + } + + credential_resolver = ScopedCredentialResolver(**resolver_kwargs) + + multisafepay_sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + + # Get the managers from the SDK + order_manager = multisafepay_sdk.get_order_manager() + event_manager = multisafepay_sdk.get_event_manager() + pos_manager = multisafepay_sdk.get_pos_manager() + + order_request = ( + OrderRequest() + .add_type("redirect") + .add_order_id(f"cloud-pos-receipt-{int(time.time())}") + .add_description("Cloud POS order for receipt") + .add_amount(100) + .add_currency("EUR") + .add_gateway_info( + { + "terminal_id": terminal_id, + }, + ) + ) + + create_response = order_manager.create( + order_request, + terminal_group_id=scoped_terminal_group_id, + ) + order = create_response.get_data() + + if order is None or not order.order_id: + raise RuntimeError("Order creation did not return order_id") + + print(f"Created Cloud POS order: {order.order_id}") + print("Waiting for completed event...") + + with event_manager.subscribe_order_events(order, timeout=45.0) as stream: + for event in stream: + print(event) + + if not _is_completed_event(event): + continue + + print("Completed event detected. Fetching receipt...") + receipt_response = pos_manager.get_receipt( + order_id=order.order_id, + terminal_group_id=scoped_terminal_group_id, + ) + print(receipt_response.get_data()) + break diff --git a/src/multisafepay/api/paths/orders/order_id/cancel/__init__.py b/src/multisafepay/api/paths/orders/order_id/cancel/__init__.py new file mode 100644 index 0000000..37e056f --- /dev/null +++ b/src/multisafepay/api/paths/orders/order_id/cancel/__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. + +"""Cancel operations and endpoints for specific orders.""" diff --git a/src/multisafepay/api/paths/orders/order_id/cancel/request/__init__.py b/src/multisafepay/api/paths/orders/order_id/cancel/request/__init__.py new file mode 100644 index 0000000..6640c14 --- /dev/null +++ b/src/multisafepay/api/paths/orders/order_id/cancel/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 order cancellation operations.""" + +from multisafepay.api.paths.orders.order_id.cancel.request.cancel_transaction_request import ( + CancelTransactionRequest, +) + +__all__ = [ + "CancelTransactionRequest", +] diff --git a/src/multisafepay/api/paths/orders/order_id/cancel/request/cancel_transaction_request.py b/src/multisafepay/api/paths/orders/order_id/cancel/request/cancel_transaction_request.py new file mode 100644 index 0000000..d477b69 --- /dev/null +++ b/src/multisafepay/api/paths/orders/order_id/cancel/request/cancel_transaction_request.py @@ -0,0 +1,42 @@ +# 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 cancel transaction endpoint.""" + +from multisafepay.model.request_model import RequestModel + + +class CancelTransactionRequest(RequestModel): + """ + Represents a request to cancel a POS transaction. + + Attributes + ---------- + order_id (str): The order identifier used in the endpoint path. + + """ + + order_id: str + + def add_order_id( + self: "CancelTransactionRequest", + order_id: str, + ) -> "CancelTransactionRequest": + """ + Adds order id to the cancellation request. + + Parameters + ---------- + order_id (str): The order identifier. + + Returns + ------- + CancelTransactionRequest: The updated request instance. + + """ + self.order_id = order_id + return self diff --git a/src/multisafepay/api/paths/orders/order_id/cancel/response/__init__.py b/src/multisafepay/api/paths/orders/order_id/cancel/response/__init__.py new file mode 100644 index 0000000..356921c --- /dev/null +++ b/src/multisafepay/api/paths/orders/order_id/cancel/response/__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. + +"""Response models for order cancellation outcomes.""" + +from multisafepay.api.paths.orders.order_id.cancel.response.cancel_transaction import ( + CancelTransaction, +) + +__all__ = [ + "CancelTransaction", +] diff --git a/src/multisafepay/api/paths/orders/order_id/cancel/response/cancel_transaction.py b/src/multisafepay/api/paths/orders/order_id/cancel/response/cancel_transaction.py new file mode 100644 index 0000000..ff77591 --- /dev/null +++ b/src/multisafepay/api/paths/orders/order_id/cancel/response/cancel_transaction.py @@ -0,0 +1,79 @@ +# 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 order cancellation endpoint payload.""" + +from typing import Optional + +from multisafepay.api.base.decorator import Decorator +from multisafepay.api.paths.orders.response.components.payment_details import ( + PaymentDetails, +) +from multisafepay.api.shared.costs import Costs +from multisafepay.api.shared.custom_info import CustomInfo +from multisafepay.api.shared.customer import Customer +from multisafepay.api.shared.payment_method import PaymentMethod +from multisafepay.model.response_model import ResponseModel + + +class CancelTransaction(ResponseModel): + """ + Represents the `data` payload returned by cancel order transaction. + + Attributes + ---------- + costs (Optional[list[Costs]]): The costs of the order. + created (Optional[str]): Creation timestamp. + modified (Optional[str]): Last modification timestamp. + custom_info (Optional[CustomInfo]): Additional custom info. + customer (Optional[Customer]): The customer data. + fastcheckout (Optional[str]): Fastcheckout flag/status. + financial_status (Optional[str]): Financial status. + items (Optional[str]): Rendered items payload. + payment_details (Optional[PaymentDetails]): Payment details. + payment_methods (Optional[list[PaymentMethod]]): Payment methods. + status (Optional[str]): Order status. + + """ + + costs: Optional[list[Costs]] + created: Optional[str] + modified: Optional[str] + custom_info: Optional[CustomInfo] + customer: Optional[Customer] + fastcheckout: Optional[str] + financial_status: Optional[str] + items: Optional[str] + payment_details: Optional[PaymentDetails] + payment_methods: Optional[list[PaymentMethod]] + status: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["CancelTransaction"]: + """ + Create a CancelTransaction from dictionary data. + + Parameters + ---------- + d (dict): The cancellation response data. + + Returns + ------- + Optional[CancelTransaction]: A cancellation response instance or None. + + """ + if d is None: + return None + cancel_dependency_adapter = Decorator(dependencies=d) + dependencies = ( + cancel_dependency_adapter.adapt_costs(d.get("costs")) + .adapt_custom_info(d.get("custom_info")) + .adapt_payment_details(d.get("payment_details")) + .adapt_payment_methods(d.get("payment_methods")) + .get_dependencies() + ) + return CancelTransaction(**dependencies) diff --git a/src/multisafepay/api/paths/orders/order_manager.py b/src/multisafepay/api/paths/orders/order_manager.py index 9ae9092..e9d9fbe 100644 --- a/src/multisafepay/api/paths/orders/order_manager.py +++ b/src/multisafepay/api/paths/orders/order_manager.py @@ -15,6 +15,12 @@ from multisafepay.api.base.response.custom_api_response import ( CustomApiResponse, ) +from multisafepay.api.paths.orders.order_id.cancel.request.cancel_transaction_request import ( + CancelTransactionRequest, +) +from multisafepay.api.paths.orders.order_id.cancel.response.cancel_transaction import ( + CancelTransaction, +) from multisafepay.api.paths.orders.order_id.capture.request.capture_request import ( CaptureOrderRequest, ) @@ -38,6 +44,7 @@ from multisafepay.api.shared.cart.shopping_cart import ShoppingCart from multisafepay.api.shared.description import Description from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope from multisafepay.util.dict_utils import dict_empty from multisafepay.util.json_encoder import DecimalEncoder from multisafepay.util.message import MessageList, gen_could_not_created_msg @@ -90,6 +97,26 @@ def __custom_api_response(response: ApiResponse) -> CustomApiResponse: return CustomApiResponse(**args) + @staticmethod + def __custom_cancel_transaction_response( + response: ApiResponse, + ) -> CustomApiResponse: + args: dict = { + **response.dict(), + "data": None, + } + if not dict_empty(response.get_body_data()): + try: + args["data"] = CancelTransaction.from_dict( + d=response.get_body_data().copy(), + ) + except ValidationError: + args["warnings"] = MessageList().add_message( + gen_could_not_created_msg("CancelTransaction"), + ) + + return CustomApiResponse(**args) + def get(self: "OrderManager", order_id: str) -> CustomApiResponse: """ Retrieve an order by its ID. @@ -115,6 +142,7 @@ def get(self: "OrderManager", order_id: str) -> CustomApiResponse: def create( self: "OrderManager", request_order: OrderRequest, + terminal_group_id: str = None, ) -> CustomApiResponse: """ Create a new order. @@ -122,6 +150,8 @@ def create( Parameters ---------- request_order (OrderRequest): The request object containing order details. + terminal_group_id (str): Optional terminal group identifier for + scoped auth resolution. Returns ------- @@ -132,6 +162,14 @@ def create( response: ApiResponse = self.client.create_post_request( "json/orders", request_body=json_data, + auth_scope=( + AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=terminal_group_id, + ) + if terminal_group_id + else None + ), ) return OrderManager.__custom_api_response(response) @@ -246,6 +284,51 @@ def refund( return CustomApiResponse(**args) + def cancel_transaction( + self: "OrderManager", + cancel_transaction_request: Union[ + CancelTransactionRequest, + str, + ], + terminal_group_id: str = None, + ) -> CustomApiResponse: + """ + Cancel a POS transaction by order id. + + Parameters + ---------- + cancel_transaction_request (CancelTransactionRequest | str): + Request object or direct order identifier. + terminal_group_id (str): Optional terminal group identifier for + scoped auth resolution. + + Returns + ------- + CustomApiResponse: The cancellation response. + + """ + order_id = ( + cancel_transaction_request + if isinstance(cancel_transaction_request, str) + else cancel_transaction_request.order_id + ) + encoded_order_id = self.encode_path_segment(order_id) + endpoint = f"json/orders/{encoded_order_id}/cancel" + context = {"order_id": order_id} + response = self.client.create_post_request( + endpoint, + context=context, + auth_scope=( + AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=terminal_group_id, + ) + if terminal_group_id + else None + ), + ) + return OrderManager.__custom_cancel_transaction_response(response) + def refund_by_item( self: "OrderManager", order: Order, diff --git a/src/multisafepay/api/paths/pos/__init__.py b/src/multisafepay/api/paths/pos/__init__.py new file mode 100644 index 0000000..3d5f3bc --- /dev/null +++ b/src/multisafepay/api/paths/pos/__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. + +"""Point of Sale API endpoints.""" + +from multisafepay.api.paths.pos.pos_manager import PosManager + +__all__ = [ + "PosManager", +] diff --git a/src/multisafepay/api/paths/pos/pos_manager.py b/src/multisafepay/api/paths/pos/pos_manager.py new file mode 100644 index 0000000..7bf3dce --- /dev/null +++ b/src/multisafepay/api/paths/pos/pos_manager.py @@ -0,0 +1,89 @@ +# 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. + +"""POS manager for `/json/pos/...` endpoints.""" + +from multisafepay.api.base.abstract_manager import AbstractManager +from multisafepay.api.base.response.api_response import ApiResponse +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.pos.receipt.response.receipt import Receipt +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 + + +class PosManager(AbstractManager): + """A class representing the PosManager.""" + + def __init__(self: "PosManager", client: Client) -> None: + """ + Initialize the PosManager with a client. + + Parameters + ---------- + client (Client): The client used to make API requests. + + """ + super().__init__(client) + + @staticmethod + def __custom_receipt_response(response: ApiResponse) -> CustomApiResponse: + args: dict = { + **response.dict(), + "data": None, + } + if not dict_empty(response.get_body_data()): + try: + args["data"] = Receipt.from_dict( + d=response.get_body_data().copy(), + ) + except ValidationError: + args["warnings"] = MessageList().add_message( + gen_could_not_created_msg("Receipt"), + ) + + return CustomApiResponse(**args) + + def get_receipt( + self: "PosManager", + order_id: str, + terminal_group_id: str = None, + ) -> CustomApiResponse: + """ + Retrieve receipt data for a POS transaction. + + Parameters + ---------- + order_id (str): Order identifier. + terminal_group_id (str): Optional terminal group identifier for + scoped auth. + + Returns + ------- + CustomApiResponse: The response containing receipt data. + + """ + encoded_order_id = self.encode_path_segment(order_id) + endpoint = f"json/pos/receipt/{encoded_order_id}" + context = {"order_id": order_id} + response = self.client.create_get_request( + endpoint, + context=context, + auth_scope=( + AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=terminal_group_id, + ) + if terminal_group_id + else None + ), + ) + return PosManager.__custom_receipt_response(response) diff --git a/src/multisafepay/api/paths/pos/receipt/__init__.py b/src/multisafepay/api/paths/pos/receipt/__init__.py new file mode 100644 index 0000000..f3c3df3 --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/__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. + +"""POS receipt endpoint models.""" diff --git a/src/multisafepay/api/paths/pos/receipt/response/__init__.py b/src/multisafepay/api/paths/pos/receipt/response/__init__.py new file mode 100644 index 0000000..a6d3fab --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/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 POS receipt endpoint.""" + +from multisafepay.api.paths.pos.receipt.response.receipt import Receipt + +__all__ = [ + "Receipt", +] diff --git a/src/multisafepay/api/paths/pos/receipt/response/components/__init__.py b/src/multisafepay/api/paths/pos/receipt/response/components/__init__.py new file mode 100644 index 0000000..e1605ad --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/components/__init__.py @@ -0,0 +1,28 @@ +# 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. + +"""Component models for POS receipt response payload.""" + +from multisafepay.api.paths.pos.receipt.response.components.merchant import ( + ReceiptMerchant, +) +from multisafepay.api.paths.pos.receipt.response.components.order import ( + ReceiptOrder, +) +from multisafepay.api.paths.pos.receipt.response.components.payment import ( + ReceiptPayment, +) +from multisafepay.api.paths.pos.receipt.response.components.related_transactions import ( + ReceiptRelatedTransactions, +) + +__all__ = [ + "ReceiptMerchant", + "ReceiptOrder", + "ReceiptPayment", + "ReceiptRelatedTransactions", +] diff --git a/src/multisafepay/api/paths/pos/receipt/response/components/merchant.py b/src/multisafepay/api/paths/pos/receipt/response/components/merchant.py new file mode 100644 index 0000000..5ee703c --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/components/merchant.py @@ -0,0 +1,26 @@ +# 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. + +"""Merchant section for POS receipt response.""" + +from typing import Optional + +from multisafepay.model.response_model import ResponseModel + + +class ReceiptMerchant(ResponseModel): + """Merchant information included in receipt data.""" + + name: Optional[str] + address: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptMerchant"]: + """Create a ReceiptMerchant model from dictionary data.""" + if d is None: + return None + return ReceiptMerchant(**d) diff --git a/src/multisafepay/api/paths/pos/receipt/response/components/order.py b/src/multisafepay/api/paths/pos/receipt/response/components/order.py new file mode 100644 index 0000000..0fd81e8 --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/components/order.py @@ -0,0 +1,109 @@ +# 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. + +"""Order section models for POS receipt response.""" + +from typing import Optional + +from multisafepay.model.response_model import ResponseModel + + +class ReceiptOrderItem(ResponseModel): + """Represents one printed order item on the receipt.""" + + currency: Optional[str] + item_price: Optional[int] + name: Optional[str] + quantity: Optional[int] + unit_price: Optional[int] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptOrderItem"]: + """Create a ReceiptOrderItem from dictionary data.""" + if d is None: + return None + return ReceiptOrderItem(**d) + + +class ReceiptOrderTipEmployee(ResponseModel): + """Employee info for receipt tip section.""" + + id: Optional[str] + name: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptOrderTipEmployee"]: + """Create a ReceiptOrderTipEmployee from dictionary data.""" + if d is None: + return None + return ReceiptOrderTipEmployee(**d) + + +class ReceiptOrderTip(ResponseModel): + """Tip information on the printed receipt.""" + + amount: Optional[int] + employee: Optional[list[ReceiptOrderTipEmployee]] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptOrderTip"]: + """Create a ReceiptOrderTip from dictionary data.""" + if d is None: + return None + + args = d.copy() + employee_data = d.get("employee") + if isinstance(employee_data, list): + args["employee"] = [ + ReceiptOrderTipEmployee.from_dict(employee) + for employee in employee_data + if isinstance(employee, dict) + ] + + return ReceiptOrderTip(**args) + + +class ReceiptOrder(ResponseModel): + """Order information included in receipt data.""" + + amount: Optional[int] + amount_refunded: Optional[int] + completed: Optional[str] + created: Optional[str] + currency: Optional[str] + financial_status: Optional[str] + modified: Optional[str] + order_id: Optional[str] + status: Optional[str] + transaction_id: Optional[int] + items: Optional[list[ReceiptOrderItem]] + tip: Optional[list[ReceiptOrderTip]] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptOrder"]: + """Create a ReceiptOrder from dictionary data.""" + if d is None: + return None + + args = d.copy() + items_data = d.get("items") + if isinstance(items_data, list): + args["items"] = [ + ReceiptOrderItem.from_dict(item) + for item in items_data + if isinstance(item, dict) + ] + + tip_data = d.get("tip") + if isinstance(tip_data, list): + args["tip"] = [ + ReceiptOrderTip.from_dict(tip) + for tip in tip_data + if isinstance(tip, dict) + ] + + return ReceiptOrder(**args) diff --git a/src/multisafepay/api/paths/pos/receipt/response/components/payment.py b/src/multisafepay/api/paths/pos/receipt/response/components/payment.py new file mode 100644 index 0000000..9a0c753 --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/components/payment.py @@ -0,0 +1,36 @@ +# 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. + +"""Payment section model for POS receipt response.""" + +from typing import Optional + +from multisafepay.model.response_model import ResponseModel + + +class ReceiptPayment(ResponseModel): + """Payment information included in receipt data.""" + + application_id: Optional[str] + authorization_code: Optional[int] + card_acceptor_location: Optional[str] + card_entry_mode: Optional[str] + card_expiry_date: Optional[str] + cardholder_verification_method: Optional[str] + issuer_bin: Optional[str] + issuer_country_code: Optional[str] + last4: Optional[str] + payment_method: Optional[str] + response_code: Optional[str] + terminal_id: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptPayment"]: + """Create a ReceiptPayment model from dictionary data.""" + if d is None: + return None + return ReceiptPayment(**d) diff --git a/src/multisafepay/api/paths/pos/receipt/response/components/related_transactions.py b/src/multisafepay/api/paths/pos/receipt/response/components/related_transactions.py new file mode 100644 index 0000000..f96d40b --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/components/related_transactions.py @@ -0,0 +1,36 @@ +# 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. + +"""Related transactions section model for POS receipt response.""" + +from typing import Optional + +from multisafepay.model.response_model import ResponseModel + + +class ReceiptRelatedTransactions(ResponseModel): + """Represents related transaction data returned for a receipt.""" + + amount: Optional[int] + created: Optional[str] + currency: Optional[str] + description: Optional[str] + items: Optional[str] + modified: Optional[str] + order_id: Optional[str] + reference_transaction_id: Optional[int] + status: Optional[str] + transaction_id: Optional[int] + type: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional["ReceiptRelatedTransactions"]: + """Create a ReceiptRelatedTransactions model from dictionary data.""" + if d is None: + return None + + return ReceiptRelatedTransactions(**d) diff --git a/src/multisafepay/api/paths/pos/receipt/response/receipt.py b/src/multisafepay/api/paths/pos/receipt/response/receipt.py new file mode 100644 index 0000000..5a2e7bd --- /dev/null +++ b/src/multisafepay/api/paths/pos/receipt/response/receipt.py @@ -0,0 +1,78 @@ +# 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 receipt data.""" + +from typing import Optional + +from multisafepay.api.paths.pos.receipt.response.components.merchant import ( + ReceiptMerchant, +) +from multisafepay.api.paths.pos.receipt.response.components.order import ( + ReceiptOrder, +) +from multisafepay.api.paths.pos.receipt.response.components.payment import ( + ReceiptPayment, +) +from multisafepay.api.paths.pos.receipt.response.components.related_transactions import ( + ReceiptRelatedTransactions, +) +from multisafepay.model.response_model import ResponseModel + + +class Receipt(ResponseModel): + """ + Represents receipt payload data returned by the POS receipt endpoint. + + Attributes + ---------- + merchant (Optional[ReceiptMerchant]): Information about the merchant. + order (Optional[ReceiptOrder]): Information about the order. + payment (Optional[ReceiptPayment]): Information about the payment. + printed_on (Optional[str]): Timestamp when the receipt was printed. + related_transactions (Optional[ReceiptRelatedTransactions]): Linked transaction information. + + """ + + merchant: Optional[ReceiptMerchant] + order: Optional[ReceiptOrder] + payment: Optional[ReceiptPayment] + printed_on: Optional[str] + related_transactions: Optional[ReceiptRelatedTransactions] + + @staticmethod + def from_dict(d: dict) -> Optional["Receipt"]: + """ + Create a Receipt from dictionary data. + + Parameters + ---------- + d (dict): The receipt data. + + Returns + ------- + Optional[Receipt]: A receipt instance or None. + + """ + if d is None: + return None + + args = d.copy() + if isinstance(d.get("merchant"), dict): + args["merchant"] = ReceiptMerchant.from_dict(d.get("merchant")) + if isinstance(d.get("order"), dict): + args["order"] = ReceiptOrder.from_dict(d.get("order")) + if isinstance(d.get("payment"), dict): + args["payment"] = ReceiptPayment.from_dict(d.get("payment")) + if isinstance(d.get("related_transactions"), dict): + args["related_transactions"] = ( + ReceiptRelatedTransactions.from_dict( + d.get("related_transactions"), + ) + ) + + return Receipt(**args) 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..4ee2474 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,9 +333,10 @@ def _create_request( ApiResponse: The API response. """ + api_key = self._resolve_api_key(auth_scope) headers = { - "Authorization": "Bearer " + self.api_key.get(), - "accept-encoding": "application/json", + "Authorization": "Bearer " + api_key, + "Accept": "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..2572035 100644 --- a/src/multisafepay/sdk.py +++ b/src/multisafepay/sdk.py @@ -17,6 +17,7 @@ from multisafepay.api.paths.payment_methods.payment_method_manager import ( PaymentMethodManager, ) +from multisafepay.api.paths.pos.pos_manager import PosManager from multisafepay.api.paths.transactions.transaction_manager import ( TransactionManager, ) @@ -26,6 +27,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 +40,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 +64,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 +198,18 @@ def get_capture_manager(self: "Sdk") -> CaptureManager: """ return CaptureManager(self.client) + def get_pos_manager(self: "Sdk") -> PosManager: + """ + Get the POS manager. + + Returns + ------- + PosManager + The POS manager instance. + + """ + return PosManager(self.client) + def get_client(self: "Sdk") -> Client: """ Get the client instance. diff --git a/tests/multisafepay/integration/api/path/orders/manager/test_integration_order_manager_create.py b/tests/multisafepay/integration/api/path/orders/manager/test_integration_order_manager_create.py index 4b14730..8942814 100644 --- a/tests/multisafepay/integration/api/path/orders/manager/test_integration_order_manager_create.py +++ b/tests/multisafepay/integration/api/path/orders/manager/test_integration_order_manager_create.py @@ -21,6 +21,8 @@ from multisafepay.api.paths.orders.request.order_request import OrderRequest from multisafepay.api.paths.orders.response.order_response import Order from multisafepay.api.shared.customer import Customer +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope def test_integration_order_manager_create_redirect(): @@ -88,4 +90,47 @@ def test_integration_order_manager_create_redirect(): assert isinstance(response, CustomApiResponse) assert isinstance(response.get_data(), Order) - assert response.get_data() == Order(**data_response) + assert response.get_data() == Order.from_dict(data_response) + + +def test_integration_order_manager_create_with_terminal_group_scope(): + """Use terminal-group auth scope when terminal_group_id is provided.""" + client = MagicMock() + client.create_post_request.return_value = ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "order_id": "cloud-pos-order", + }, + }, + ) + order_request = ( + OrderRequest() + .add_type("direct") + .add_order_id("cloud-pos-order") + .add_currency("EUR") + .add_amount(100) + ) + + order_manager = OrderManager(client) + response = order_manager.create( + request_order=order_request, + terminal_group_id="Default", + ) + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Order) + assert response.get_data().order_id == "cloud-pos-order" + + called_endpoint = client.create_post_request.call_args.args[0] + called_auth_scope = client.create_post_request.call_args.kwargs[ + "auth_scope" + ] + + assert called_endpoint == "json/orders" + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id="Default", + ) diff --git a/tests/multisafepay/unit/api/path/orders/manager/test_unit_cancel_transaction.py b/tests/multisafepay/unit/api/path/orders/manager/test_unit_cancel_transaction.py new file mode 100644 index 0000000..ff39274 --- /dev/null +++ b/tests/multisafepay/unit/api/path/orders/manager/test_unit_cancel_transaction.py @@ -0,0 +1,119 @@ +# 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 OrderManager.cancel_transaction 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.orders.order_id.cancel.request.cancel_transaction_request import ( + CancelTransactionRequest, +) +from multisafepay.api.paths.orders.order_id.cancel.response.cancel_transaction import ( + CancelTransaction, +) +from multisafepay.api.paths.orders.order_manager import OrderManager +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope + +ORDER_ID = "cloud-pos-cancel-1" +TERMINAL_GROUP_ID = "Default" + + +def _build_cancel_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "status": "void", + "financial_status": "void", + "created": "2026-01-01T00:00:00", + "modified": "2026-01-01T00:00:01", + }, + }, + ) + + +def test_cancel_transaction_with_terminal_group_scope() -> None: + """Use terminal-group auth scope when terminal_group_id is provided.""" + client = MagicMock() + client.create_post_request.return_value = _build_cancel_api_response() + + manager = OrderManager(client) + response = manager.cancel_transaction( + cancel_transaction_request=ORDER_ID, + terminal_group_id=TERMINAL_GROUP_ID, + ) + + called_auth_scope = client.create_post_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), CancelTransaction) + assert response.get_data().status == "void" + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=TERMINAL_GROUP_ID, + ) + + +def test_cancel_transaction_without_terminal_group_scope() -> None: + """Omit auth scope when terminal_group_id is not provided.""" + client = MagicMock() + client.create_post_request.return_value = _build_cancel_api_response() + + manager = OrderManager(client) + response = manager.cancel_transaction( + cancel_transaction_request=ORDER_ID, + ) + + called_auth_scope = client.create_post_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert called_auth_scope is None + + +def test_cancel_transaction_accepts_request_object() -> None: + """Accept CancelTransactionRequest as input instead of raw string.""" + client = MagicMock() + client.create_post_request.return_value = _build_cancel_api_response() + + request = CancelTransactionRequest(order_id=ORDER_ID) + + manager = OrderManager(client) + response = manager.cancel_transaction( + cancel_transaction_request=request, + terminal_group_id=TERMINAL_GROUP_ID, + ) + + called_endpoint = client.create_post_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert ORDER_ID in called_endpoint + assert called_endpoint.endswith("/cancel") + + +def test_cancel_transaction_encodes_order_id() -> None: + """Verify order ID with special chars is encoded in the endpoint.""" + client = MagicMock() + client.create_post_request.return_value = _build_cancel_api_response() + + manager = OrderManager(client) + manager.cancel_transaction( + cancel_transaction_request="order/special&chars", + ) + + called_endpoint = client.create_post_request.call_args.args[0] + assert "order%2Fspecial%26chars" in called_endpoint diff --git a/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager.py b/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager.py new file mode 100644 index 0000000..50eba9f --- /dev/null +++ b/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager.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. + +"""Unit tests for basic order manager create 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.orders.order_manager import OrderManager +from multisafepay.api.paths.orders.request.order_request import OrderRequest +from multisafepay.api.paths.orders.response.order_response import Order +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope + +ORDERS_ENDPOINT = "json/orders" +TERMINAL_GROUP_ID = "Default" +SCOPED_ORDER_ID = "cloud-pos-order" +DEFAULT_ORDER_ID = "default-order" + + +def _build_api_response(order_id: str) -> ApiResponse: + """Create a minimal successful ApiResponse for order manager tests.""" + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "order_id": order_id, + }, + }, + ) + + +def _build_order_request(order_id: str) -> OrderRequest: + """Create a minimal valid order request used in create() tests.""" + return ( + OrderRequest() + .add_type("direct") + .add_order_id(order_id) + .add_currency("EUR") + .add_amount(100) + ) + + +def test_create_uses_terminal_group_auth_scope_when_provided() -> None: + """Use terminal-group scope only when terminal_group_id is passed.""" + client = MagicMock() + client.create_post_request.return_value = _build_api_response( + SCOPED_ORDER_ID, + ) + request_order = _build_order_request(SCOPED_ORDER_ID) + + manager = OrderManager(client) + response = manager.create( + request_order=request_order, + terminal_group_id=TERMINAL_GROUP_ID, + ) + + called_endpoint = client.create_post_request.call_args.args[0] + called_auth_scope = client.create_post_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Order) + assert called_endpoint == ORDERS_ENDPOINT + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=TERMINAL_GROUP_ID, + ) + + +def test_create_omits_auth_scope_when_terminal_group_id_is_not_passed() -> ( + None +): + """Do not set auth_scope when create request has no terminal group id.""" + client = MagicMock() + client.create_post_request.return_value = _build_api_response( + DEFAULT_ORDER_ID, + ) + request_order = _build_order_request(DEFAULT_ORDER_ID) + + manager = OrderManager(client) + response = manager.create(request_order=request_order) + + called_endpoint = client.create_post_request.call_args.args[0] + called_auth_scope = client.create_post_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Order) + assert response.get_data().order_id == DEFAULT_ORDER_ID + assert called_endpoint == ORDERS_ENDPOINT + assert called_auth_scope is None diff --git a/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager_operations.py b/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager_operations.py new file mode 100644 index 0000000..c6135d4 --- /dev/null +++ b/tests/multisafepay/unit/api/path/orders/manager/test_unit_order_manager_operations.py @@ -0,0 +1,170 @@ +# 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 OrderManager get, update, capture, and refund methods.""" + +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.orders.order_manager import OrderManager +from multisafepay.api.paths.orders.response.order_response import Order + +ORDER_ID = "test-order-1" + + +def _build_order_api_response(order_id: str) -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "order_id": order_id, + "status": "completed", + }, + }, + ) + + +def _build_empty_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={"success": True, "data": {}}, + ) + + +def _build_capture_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "order_id": ORDER_ID, + "status": "completed", + }, + }, + ) + + +def _build_refund_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "refund_id": "refund-1", + "order_id": ORDER_ID, + }, + }, + ) + + +def test_get_order_by_id() -> None: + """Retrieve an order by its ID.""" + client = MagicMock() + client.create_get_request.return_value = _build_order_api_response( + ORDER_ID, + ) + + manager = OrderManager(client) + response = manager.get(order_id=ORDER_ID) + + called_endpoint = client.create_get_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Order) + assert response.get_data().order_id == ORDER_ID + assert ORDER_ID in called_endpoint + + +def test_get_order_encodes_special_chars_in_id() -> None: + """Verify order ID with special chars is encoded in the URL.""" + client = MagicMock() + client.create_get_request.return_value = _build_order_api_response( + "order/special", + ) + + manager = OrderManager(client) + manager.get(order_id="order/special") + + called_endpoint = client.create_get_request.call_args.args[0] + assert "order%2Fspecial" in called_endpoint + + +def test_get_order_returns_none_for_empty_data() -> None: + """Return None when body data is empty.""" + client = MagicMock() + client.create_get_request.return_value = _build_empty_api_response() + + manager = OrderManager(client) + response = manager.get(order_id=ORDER_ID) + + assert isinstance(response, CustomApiResponse) + assert response.get_data() is None + + +def test_update_order_sends_patch_request() -> None: + """Update sends PATCH to the correct endpoint.""" + client = MagicMock() + client.create_patch_request.return_value = _build_empty_api_response() + + update_request = MagicMock() + update_request.to_dict.return_value = {"description": "updated"} + + manager = OrderManager(client) + response = manager.update(order_id=ORDER_ID, update_request=update_request) + + called_endpoint = client.create_patch_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert ORDER_ID in called_endpoint + + +def test_capture_order_sends_post_and_parses_response() -> None: + """Capture sends POST and parses OrderCapture response.""" + client = MagicMock() + client.create_post_request.return_value = _build_capture_api_response() + + capture_request = MagicMock() + capture_request.to_dict.return_value = {"amount": 100} + + manager = OrderManager(client) + response = manager.capture( + order_id=ORDER_ID, + capture_request=capture_request, + ) + + called_endpoint = client.create_post_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert f"json/orders/{ORDER_ID}/capture" == called_endpoint + + +def test_refund_order_sends_post_and_parses_response() -> None: + """Refund sends POST and parses OrderRefund response.""" + client = MagicMock() + client.create_post_request.return_value = _build_refund_api_response() + + refund_request = MagicMock() + refund_request.to_dict.return_value = {"amount": 50} + + manager = OrderManager(client) + response = manager.refund( + order_id=ORDER_ID, + request_refund=refund_request, + ) + + called_endpoint = client.create_post_request.call_args.args[0] + + assert isinstance(response, CustomApiResponse) + assert f"json/orders/{ORDER_ID}/refunds" == called_endpoint diff --git a/tests/multisafepay/unit/api/path/pos/__init__.py b/tests/multisafepay/unit/api/path/pos/__init__.py new file mode 100644 index 0000000..9383b1b --- /dev/null +++ b/tests/multisafepay/unit/api/path/pos/__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 POS path package.""" diff --git a/tests/multisafepay/unit/api/path/pos/test_unit_pos_manager.py b/tests/multisafepay/unit/api/path/pos/test_unit_pos_manager.py new file mode 100644 index 0000000..0732780 --- /dev/null +++ b/tests/multisafepay/unit/api/path/pos/test_unit_pos_manager.py @@ -0,0 +1,194 @@ +# 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 PosManager.get_receipt 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.pos.pos_manager import PosManager +from multisafepay.api.paths.pos.receipt.response.receipt import Receipt +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope + +ORDER_ID = "cloud-pos-order-1" +TERMINAL_GROUP_ID = "Default" + + +def _build_receipt_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={ + "success": True, + "data": { + "merchant": {"name": "Test Merchant", "address": "123 St"}, + "order": { + "order_id": ORDER_ID, + "amount": 100, + "currency": "EUR", + "status": "completed", + "financial_status": "completed", + "created": "2026-01-01T00:00:00", + "modified": "2026-01-01T00:00:01", + "completed": "2026-01-01T00:00:02", + "amount_refunded": 0, + "transaction_id": 12345, + "items": [ + { + "name": "Widget", + "quantity": 1, + "unit_price": 100, + "item_price": 100, + "currency": "EUR", + }, + ], + "tip": [ + { + "amount": 50, + "employee": [ + {"id": "emp-1", "name": "Alice"}, + ], + }, + ], + }, + "payment": { + "payment_method": "VISA", + "last4": "1234", + "terminal_id": "T-001", + "authorization_code": 123456, + "application_id": "A001", + "card_acceptor_location": "NL", + "card_entry_mode": "contactless", + "card_expiry_date": "12/28", + "cardholder_verification_method": "pin", + "issuer_bin": "411111", + "issuer_country_code": "NL", + "response_code": "00", + }, + "printed_on": "2026-01-01T00:00:03", + "related_transactions": { + "amount": 100, + "created": "2026-01-01T00:00:00", + "currency": "EUR", + "description": "Refund", + "items": None, + "modified": "2026-01-01T00:00:01", + "order_id": ORDER_ID, + "reference_transaction_id": 99999, + "status": "completed", + "transaction_id": 12346, + "type": "refund", + }, + }, + }, + ) + + +def _build_empty_receipt_api_response() -> ApiResponse: + return ApiResponse( + headers={}, + status_code=200, + body={"success": True, "data": {}}, + ) + + +def test_get_receipt_with_terminal_group_scope() -> None: + """Use terminal-group auth scope when terminal_group_id is provided.""" + client = MagicMock() + client.create_get_request.return_value = _build_receipt_api_response() + + manager = PosManager(client) + response = manager.get_receipt( + order_id=ORDER_ID, + terminal_group_id=TERMINAL_GROUP_ID, + ) + + called_auth_scope = client.create_get_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Receipt) + assert response.get_data().printed_on == "2026-01-01T00:00:03" + assert called_auth_scope == AuthScope( + scope=Client.AUTH_SCOPE_TERMINAL_GROUP, + group_id=TERMINAL_GROUP_ID, + ) + + +def test_get_receipt_without_terminal_group_scope() -> None: + """Omit auth scope when terminal_group_id is not provided.""" + client = MagicMock() + client.create_get_request.return_value = _build_receipt_api_response() + + manager = PosManager(client) + response = manager.get_receipt(order_id=ORDER_ID) + + called_auth_scope = client.create_get_request.call_args.kwargs[ + "auth_scope" + ] + + assert isinstance(response, CustomApiResponse) + assert isinstance(response.get_data(), Receipt) + assert called_auth_scope is None + + +def test_get_receipt_parses_nested_receipt_components() -> None: + """Verify receipt response parses all nested model components.""" + client = MagicMock() + client.create_get_request.return_value = _build_receipt_api_response() + + manager = PosManager(client) + response = manager.get_receipt( + order_id=ORDER_ID, + terminal_group_id=TERMINAL_GROUP_ID, + ) + receipt = response.get_data() + + assert receipt.merchant.name == "Test Merchant" + assert receipt.merchant.address == "123 St" + assert receipt.order.order_id == ORDER_ID + assert receipt.order.amount == 100 + assert len(receipt.order.items) == 1 + assert receipt.order.items[0].name == "Widget" + assert len(receipt.order.tip) == 1 + assert receipt.order.tip[0].amount == 50 + assert receipt.order.tip[0].employee[0].name == "Alice" + assert receipt.payment.payment_method == "VISA" + assert receipt.payment.last4 == "1234" + assert receipt.related_transactions.transaction_id == 12346 + assert receipt.related_transactions.type == "refund" + + +def test_get_receipt_returns_none_data_for_empty_body() -> None: + """Return None data when body data is empty.""" + client = MagicMock() + client.create_get_request.return_value = ( + _build_empty_receipt_api_response() + ) + + manager = PosManager(client) + response = manager.get_receipt(order_id=ORDER_ID) + + assert isinstance(response, CustomApiResponse) + assert response.get_data() is None + + +def test_get_receipt_encodes_order_id_in_endpoint() -> None: + """Verify order ID with special chars is encoded in the URL.""" + client = MagicMock() + client.create_get_request.return_value = _build_receipt_api_response() + + manager = PosManager(client) + manager.get_receipt(order_id="order/special&chars") + + called_endpoint = client.create_get_request.call_args.args[0] + assert "order%2Fspecial%26chars" in called_endpoint 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..f160905 100644 --- a/tests/multisafepay/unit/test_unit_sdk.py +++ b/tests/multisafepay/unit/test_unit_sdk.py @@ -11,6 +11,38 @@ from multisafepay import Sdk 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 +96,44 @@ 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}"