diff --git a/examples/event_manager/subscribe_events.py b/examples/event_manager/subscribe_events.py new file mode 100644 index 0000000..826fa89 --- /dev/null +++ b/examples/event_manager/subscribe_events.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. + +"""Create a Cloud POS order and subscribe to its event stream.""" + +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 = _require_first_env( + "CLOUD_POS_TERMINAL_GROUP_ID", +) +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() + event_manager = multisafepay_sdk.get_event_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}") + print("Listening for events. Press Ctrl+C to stop.") + + try: + with event_manager.subscribe_order_events(order, timeout=45.0) as stream: + for event in stream: + print(event) + except KeyboardInterrupt: + print("Stream interrupted by user.") 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/events/__init__.py b/src/multisafepay/api/paths/events/__init__.py new file mode 100644 index 0000000..cf12cf7 --- /dev/null +++ b/src/multisafepay/api/paths/events/__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. + +"""Events API endpoints.""" + +from multisafepay.api.paths.events.event_manager import EventManager + +__all__ = [ + "EventManager", +] diff --git a/src/multisafepay/api/paths/events/event_manager.py b/src/multisafepay/api/paths/events/event_manager.py new file mode 100644 index 0000000..8cd490a --- /dev/null +++ b/src/multisafepay/api/paths/events/event_manager.py @@ -0,0 +1,87 @@ +# 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. + +"""Event manager for event stream subscription helpers.""" + +from __future__ import annotations + +from multisafepay.api.base.abstract_manager import AbstractManager +from multisafepay.api.paths.events.stream import EventStream +from multisafepay.api.paths.orders.response.order_response import Order +from multisafepay.client.client import Client + + +class EventManager(AbstractManager): + """Manages event stream subscriptions for order events.""" + + def __init__(self: EventManager, client: Client) -> None: + """Initialize the EventManager with a client.""" + super().__init__(client) + + def subscribe_events( + self: EventManager, + events_token: str, + events_stream_url: str, + last_event_id: str | None = None, + timeout: float = 30.0, + ) -> EventStream: + """ + Subscribe to order events using the SSE stream endpoint. + + Parameters + ---------- + events_token (str): Token returned by order creation for event auth. + events_stream_url (str): Full SSE stream URL. + last_event_id (str | None): Optional resume cursor. + timeout (float): Socket timeout in seconds. + + Returns + ------- + EventStream: An iterator over incoming SSE messages. + + """ + return EventStream.open( + events_token=events_token, + events_stream_url=events_stream_url, + last_event_id=last_event_id, + timeout=timeout, + ) + + def subscribe_order_events( + self: EventManager, + order: Order, + last_event_id: str | None = None, + timeout: float = 30.0, + ) -> EventStream: + """ + Subscribe to events for an existing order response object. + + Parameters + ---------- + order (Order): Order response that contains event credentials. + last_event_id (str | None): Optional resume cursor. + timeout (float): Socket timeout in seconds. + + Returns + ------- + EventStream: An iterator over incoming SSE messages. + + """ + events_token = order.events_token or order.event_token + events_stream_url = order.events_stream_url or order.event_stream_url + + if not events_token or not events_stream_url: + raise ValueError( + "Order does not contain events_token/events_stream_url.", + ) + + return self.subscribe_events( + events_token=events_token, + events_stream_url=events_stream_url, + last_event_id=last_event_id, + timeout=timeout, + ) diff --git a/src/multisafepay/api/paths/events/stream/__init__.py b/src/multisafepay/api/paths/events/stream/__init__.py new file mode 100644 index 0000000..84b2268 --- /dev/null +++ b/src/multisafepay/api/paths/events/stream/__init__.py @@ -0,0 +1,132 @@ +# 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. + +"""Event stream contracts for the events path.""" + +from __future__ import annotations + +import json + +from multisafepay.api.paths.events.stream.response import Event, EventData +from multisafepay.client.sse import ( + ServerSentEvent, + ServerSentEventStream, + StreamingResponse, +) +from typing_extensions import Self + + +def _deserialize_event_payload(raw_payload: str | None) -> object | None: + """Parse raw SSE data for this path and fall back to plain text.""" + if raw_payload is None: + return None + + try: + return json.loads(raw_payload) + except json.JSONDecodeError: + return raw_payload + + +def _to_event(server_sent_event: ServerSentEvent) -> Event: + """Adapt a generic SSE message into the events-path response model.""" + event = Event.from_dict( + { + "event": server_sent_event.event, + "data": _deserialize_event_payload(server_sent_event.data), + "event_id": server_sent_event.event_id, + "retry": server_sent_event.retry, + "raw_data": server_sent_event.raw_data, + }, + ) + if event is None: + raise ValueError("Unable to adapt SSE payload to Event.") + return event + + +class EventStream: + """Iterator over events received from the MultiSafepay SSE endpoint.""" + + def __init__( + self: EventStream, + response: StreamingResponse | None = None, + stream: ServerSentEventStream | None = None, + ) -> None: + """Initialize the stream from an HTTP response or generic SSE stream.""" + if stream is not None: + self._stream = stream + return + + if response is None: + raise ValueError( + "response is required when stream is not provided.", + ) + + self._stream = ServerSentEventStream(response=response) + + @classmethod + def _from_stream( + cls: type[EventStream], + stream: ServerSentEventStream, + ) -> EventStream: + """Build an EventStream around an already-open generic SSE stream.""" + return cls(stream=stream) + + @classmethod + def open( + cls: type[EventStream], + events_token: str, + events_stream_url: str, + last_event_id: str | None = None, + timeout: float = 30.0, + ) -> EventStream: + """Open a new SSE stream using the event token and stream URL.""" + headers = { + "Accept": "text/event-stream", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {events_token}", + } + if last_event_id is not None: + headers["Last-Event-ID"] = last_event_id + + stream = ServerSentEventStream.open( + url=events_stream_url, + headers=headers, + timeout=timeout, + ) + return cls._from_stream(stream) + + @property + def closed(self: EventStream) -> bool: + """Return whether this stream is already closed.""" + return self._stream.closed + + def close(self: EventStream) -> None: + """Close the underlying HTTP response stream.""" + self._stream.close() + + def __iter__(self: EventStream) -> EventStream: + """Return self as an iterator over events.""" + return self + + def __next__(self: EventStream) -> Event: + """Read the next SSE message and return it as an Event.""" + return _to_event(next(self._stream)) + + def __enter__(self: Self) -> Self: + """Support context manager protocol.""" + return self + + def __exit__(self: EventStream, *args: object) -> None: + """Close stream when exiting context manager.""" + self.close() + + +__all__ = [ + "Event", + "EventData", + "EventStream", +] diff --git a/src/multisafepay/api/paths/events/stream/response/__init__.py b/src/multisafepay/api/paths/events/stream/response/__init__.py new file mode 100644 index 0000000..bd51138 --- /dev/null +++ b/src/multisafepay/api/paths/events/stream/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 event stream payloads.""" + +from multisafepay.api.paths.events.stream.response.components import EventData +from multisafepay.api.paths.events.stream.response.event import Event + +__all__ = [ + "Event", + "EventData", +] diff --git a/src/multisafepay/api/paths/events/stream/response/components/__init__.py b/src/multisafepay/api/paths/events/stream/response/components/__init__.py new file mode 100644 index 0000000..9b13bc9 --- /dev/null +++ b/src/multisafepay/api/paths/events/stream/response/components/__init__.py @@ -0,0 +1,18 @@ +# 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 event stream response payloads.""" + +from multisafepay.api.paths.events.stream.response.components.event_data import ( + EventData, + EventDataPayload, +) + +__all__ = [ + "EventData", + "EventDataPayload", +] diff --git a/src/multisafepay/api/paths/events/stream/response/components/event_data.py b/src/multisafepay/api/paths/events/stream/response/components/event_data.py new file mode 100644 index 0000000..67eedb1 --- /dev/null +++ b/src/multisafepay/api/paths/events/stream/response/components/event_data.py @@ -0,0 +1,55 @@ +# 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 response models for event stream payloads.""" + +from __future__ import annotations + +from typing import Optional, Union + +from multisafepay.model.response_model import ResponseModel + +# ruff: noqa: UP007 + +EventDataPayload = Union[ + "EventData", + str, + int, + float, + bool, + list[object], + None, +] + + +class EventData(ResponseModel): + """Structured nested event payload for known order-event attributes.""" + + status: Optional[str] + order_id: Optional[str] + type: Optional[str] + data: EventDataPayload + + @staticmethod + def from_dict(d: dict) -> Optional[EventData]: + """Create EventData from dictionary data.""" + if d is None: + return None + + args = d.copy() + payload = d.get("data") + if isinstance(payload, dict): + args["data"] = EventData.from_dict(payload) + elif isinstance(payload, list): + args["data"] = [ + EventData.from_dict(item) if isinstance(item, dict) else item + for item in payload + ] + else: + args["data"] = payload + + return EventData(**args) diff --git a/src/multisafepay/api/paths/events/stream/response/event.py b/src/multisafepay/api/paths/events/stream/response/event.py new file mode 100644 index 0000000..52af229 --- /dev/null +++ b/src/multisafepay/api/paths/events/stream/response/event.py @@ -0,0 +1,50 @@ +# 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 event stream messages.""" + +from __future__ import annotations + +from typing import Optional + +from multisafepay.api.paths.events.stream.response.components import ( + EventData, + EventDataPayload, +) +from multisafepay.model.response_model import ResponseModel + +# ruff: noqa: UP007 + + +class Event(ResponseModel): + """Structured event message returned by the MultiSafepay SSE stream.""" + + event: Optional[str] + data: EventDataPayload + event_id: Optional[str] + retry: Optional[int] + raw_data: Optional[str] + + @staticmethod + def from_dict(d: dict) -> Optional[Event]: + """Create Event from dictionary data.""" + if d is None: + return None + + args = d.copy() + payload = d.get("data") + if isinstance(payload, dict): + args["data"] = EventData.from_dict(payload) + elif isinstance(payload, list): + args["data"] = [ + EventData.from_dict(item) if isinstance(item, dict) else item + for item in payload + ] + else: + args["data"] = payload + + return Event(**args) 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/orders/response/order_response.py b/src/multisafepay/api/paths/orders/response/order_response.py index 4581929..89e8f04 100644 --- a/src/multisafepay/api/paths/orders/response/order_response.py +++ b/src/multisafepay/api/paths/orders/response/order_response.py @@ -103,6 +103,11 @@ class Order(ResponseModel): payment_url: Optional[str] cancel_url: Optional[str] session_id: Optional[str] + events_token: Optional[str] + events_url: Optional[str] + events_stream_url: Optional[str] + + # Backward compatibility aliases for older API payloads. event_token: Optional[str] event_url: Optional[str] event_stream_url: Optional[str] @@ -118,6 +123,23 @@ def get_order_id(self: "Order") -> str: """ return self.order_id + @staticmethod + def _normalize_event_fields(d: dict) -> dict: + """Normalize singular/plural event keys for compatibility.""" + mapping = [ + ("events_token", "event_token"), + ("events_url", "event_url"), + ("events_stream_url", "event_stream_url"), + ] + + for plural_key, singular_key in mapping: + if d.get(plural_key) is None and d.get(singular_key) is not None: + d[plural_key] = d[singular_key] + if d.get(singular_key) is None and d.get(plural_key) is not None: + d[singular_key] = d[plural_key] + + return d + @staticmethod def from_dict(d: dict) -> Optional["Order"]: """ @@ -134,7 +156,10 @@ def from_dict(d: dict) -> Optional["Order"]: """ if d is None: return None - order_dependency_adapter = Decorator(dependencies=d) + normalized_dependencies = Order._normalize_event_fields(d.copy()) + order_dependency_adapter = Decorator( + dependencies=normalized_dependencies, + ) dependencies = ( order_dependency_adapter.adapt_order_adjustment( d.get("order_adjustment"), 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..30fdbf6 100644 --- a/src/multisafepay/client/__init__.py +++ b/src/multisafepay/client/__init__.py @@ -2,8 +2,13 @@ from multisafepay.client.api_key import ApiKey from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import ScopedCredentialResolver +from multisafepay.client.sse import ServerSentEvent, ServerSentEventStream __all__ = [ "ApiKey", "Client", + "ScopedCredentialResolver", + "ServerSentEvent", + "ServerSentEventStream", ] 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/client/sse.py b/src/multisafepay/client/sse.py new file mode 100644 index 0000000..16c0586 --- /dev/null +++ b/src/multisafepay/client/sse.py @@ -0,0 +1,224 @@ +# 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. + +"""Generic Server-Sent Events support utilities.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import ClassVar, Protocol +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +from typing_extensions import Self + + +class StreamingResponse(Protocol): + """Protocol for the minimal stream response interface used by SSE streams.""" + + def readline(self: StreamingResponse) -> bytes: + """Read one line from the stream response.""" + + def close(self: StreamingResponse) -> None: + """Close the stream response.""" + + +@dataclass(frozen=True) +class ServerSentEvent: + """Generic representation of one SSE message.""" + + event: str | None = None + data: str | None = None + event_id: str | None = None + retry: int | None = None + raw_data: str | None = None + + +@dataclass +class _ServerSentEventBuilder: + """Mutable builder used while parsing one SSE message.""" + + _FIELD_HANDLERS: ClassVar[dict[str, str]] = { + "event": "_consume_event", + "data": "_consume_data", + "id": "_consume_id", + "retry": "_consume_retry", + } + + event_name: str | None = None + event_id: str | None = None + event_retry: int | None = None + data_lines: list[str] = field(default_factory=list) + has_fields: bool = False + + def consume_line(self: _ServerSentEventBuilder, line: str) -> None: + """Consume one SSE line and update the builder state.""" + if line.startswith(":"): + return + + field_name, field_value = _parse_line(line) + if field_name is None: + return + + handler_name = self._FIELD_HANDLERS.get(field_name) + if handler_name is None: + return + + getattr(self, handler_name)(field_value) + + def _consume_event( + self: _ServerSentEventBuilder, + field_value: str, + ) -> None: + """Consume the SSE event field.""" + self.has_fields = True + self.event_name = field_value + + def _consume_data(self: _ServerSentEventBuilder, field_value: str) -> None: + """Consume one SSE data line.""" + self.has_fields = True + self.data_lines.append(field_value) + + def _consume_id(self: _ServerSentEventBuilder, field_value: str) -> None: + """Consume the SSE id field.""" + self.has_fields = True + self.event_id = field_value + + def _consume_retry( + self: _ServerSentEventBuilder, + field_value: str, + ) -> None: + """Consume the SSE retry field when it is a valid integer.""" + try: + self.event_retry = int(field_value) + except ValueError: + return + + self.has_fields = True + + def build(self: _ServerSentEventBuilder) -> ServerSentEvent | None: + """Build an immutable SSE message or None when no message exists.""" + if not self.has_fields and not self.data_lines: + return None + + raw_data = "\n".join(self.data_lines) if self.data_lines else None + return ServerSentEvent( + event=self.event_name, + data=raw_data, + event_id=self.event_id, + retry=self.event_retry, + raw_data=raw_data, + ) + + +class ServerSentEventStream: + """Iterator over messages received from a generic SSE endpoint.""" + + def __init__( + self: ServerSentEventStream, + response: StreamingResponse, + ) -> None: + """Initialize the stream from an already-open HTTP response.""" + self._response = response + self._closed = False + + @classmethod + def open( + cls: type[ServerSentEventStream], + url: str, + headers: dict[str, str] | None = None, + timeout: float = 30.0, + ) -> ServerSentEventStream: + """Open a new SSE stream using a URL and optional headers.""" + cls._validate_url(url) + + request = Request( # noqa: S310 + url=url, + headers=headers or {}, + method="GET", + ) + # Keep the response open; close manages the lifecycle. + # pylint: disable=consider-using-with + response = urlopen(request, timeout=timeout) # noqa: S310 + # pylint: enable=consider-using-with + + return cls(response=response) + + @staticmethod + def _validate_url(url: str) -> None: + """Validate the stream URL before opening the network connection.""" + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("Invalid SSE URL.") + + @property + def closed(self: ServerSentEventStream) -> bool: + """Return whether this stream is already closed.""" + return self._closed + + def close(self: ServerSentEventStream) -> None: + """Close the underlying HTTP response stream.""" + if self._closed: + return + + self._response.close() + self._closed = True + + def __iter__(self: ServerSentEventStream) -> ServerSentEventStream: + """Return self as an iterator over SSE messages.""" + return self + + def __next__(self: ServerSentEventStream) -> ServerSentEvent: + """Read the next SSE message and return it.""" + if self._closed: + raise StopIteration + + builder = _ServerSentEventBuilder() + while True: + line = self._read_line() + if line is None: + self.close() + raise StopIteration + + if line == "": + event = builder.build() + if event is not None: + return event + builder = _ServerSentEventBuilder() + continue + + builder.consume_line(line) + + def _read_line(self: ServerSentEventStream) -> str | None: + """Read and decode one line from the underlying stream response.""" + raw_line = self._response.readline() + if not raw_line: + return None + + return raw_line.decode("utf-8", errors="replace").rstrip("\r\n") + + def __enter__(self: Self) -> Self: + """Support context manager protocol.""" + return self + + def __exit__(self: ServerSentEventStream, *args: object) -> None: + """Close stream when exiting context manager.""" + self.close() + + +def _parse_line(line: str) -> tuple[str | None, str]: + """Parse one SSE line into field and value parts.""" + if ":" not in line: + return line or None, "" + + field_name, field_value = line.split(":", 1) + if field_name == "": + return None, "" + if field_value.startswith(" "): + field_value = field_value[1:] + + return field_name, field_value diff --git a/src/multisafepay/sdk.py b/src/multisafepay/sdk.py index b22dc3b..af2151e 100644 --- a/src/multisafepay/sdk.py +++ b/src/multisafepay/sdk.py @@ -11,12 +11,14 @@ from multisafepay.api.paths.auth.auth_manager import AuthManager from multisafepay.api.paths.categories.category_manager import CategoryManager +from multisafepay.api.paths.events.event_manager import EventManager from multisafepay.api.paths.gateways.gateway_manager import GatewayManager from multisafepay.api.paths.issuers.issuer_manager import IssuerManager from multisafepay.api.paths.orders.order_manager import OrderManager 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 +28,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 +41,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 +65,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) @@ -167,6 +175,18 @@ def get_category_manager(self: "Sdk") -> CategoryManager: """ return CategoryManager(self.client) + def get_event_manager(self: "Sdk") -> EventManager: + """ + Get the event manager. + + Returns + ------- + EventManager + The event manager instance. + + """ + return EventManager(self.client) + def get_order_manager(self: "Sdk") -> OrderManager: """ Get the order manager. @@ -191,6 +211,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/e2e/examples/event_manager/conftest.py b/tests/multisafepay/e2e/examples/event_manager/conftest.py new file mode 100644 index 0000000..8fb1792 --- /dev/null +++ b/tests/multisafepay/e2e/examples/event_manager/conftest.py @@ -0,0 +1,211 @@ +"""Event-manager-specific E2E fixtures and selective skip behavior.""" + +import os +from urllib.parse import urlparse + +import pytest + +from multisafepay.client import ScopedCredentialResolver +from multisafepay.client.client import Client +from multisafepay.client.credential_resolver import AuthScope +from multisafepay.sdk import Sdk + +EVENT_MANAGER_E2E_NODE_PREFIXES = ( + "tests/multisafepay/e2e/examples/event_manager/", +) +DEFAULT_API_KEY_ENVS = ("E2E_API_KEY", "API_KEY") +PARTNER_API_KEY_ENVS = ("E2E_PARTNER_API_KEY", "PARTNER_API_KEY") +TERMINAL_GROUP_API_KEY_ENVS = ( + "E2E_TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", + "TERMINAL_GROUP_API_KEY_GROUP_DEFAULT", +) +TERMINAL_GROUP_ID_ENVS = ("CLOUD_POS_TERMINAL_GROUP_ID",) +TERMINAL_ID_ENVS = ("E2E_CLOUD_POS_TERMINAL_ID", "CLOUD_POS_TERMINAL_ID") +BASE_URL_ENVS = ("E2E_NO_SANDBOX_BASE_URL", "MSP_SDK_CUSTOM_BASE_URL") + + +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 + + msg = f"SSE E2E tests require one of: {', '.join(names)}" + raise pytest.UsageError(msg) + + +def _validate_base_url(base_url: str, *env_names: str) -> str: + parsed = urlparse(base_url) + if parsed.scheme != "https" or not parsed.netloc: + msg = f"{', '.join(env_names)} 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 _has_event_manager_e2e_env() -> bool: + has_base = bool( + _get_first_env(*DEFAULT_API_KEY_ENVS) + and _get_first_env(*TERMINAL_GROUP_API_KEY_ENVS) + and _get_first_env(*TERMINAL_ID_ENVS) + and _get_first_env(*BASE_URL_ENVS), + ) + if not has_base: + return False + + return bool( + _get_first_env(*TERMINAL_GROUP_ID_ENVS) + or _get_first_env(*PARTNER_API_KEY_ENVS), + ) + + +def _resolve_terminal_group_id( + base_url: str, + default_api_key: str, + partner_api_key: str, + terminal_id: str, +) -> str: + credential_resolver = ScopedCredentialResolver( + default_api_key=default_api_key, + partner_affiliate_api_key=partner_api_key, + ) + sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + sdk.get_client().url = base_url + + limit = 100 + max_pages = 10 + for page in range(1, max_pages + 1): + response = sdk.get_client().create_get_request( + "json/terminals", + params={ + "limit": limit, + "page": page, + }, + auth_scope=AuthScope( + scope=Client.AUTH_SCOPE_PARTNER_AFFILIATE, + ), + ) + if ( + response.get_status_code() != 200 + or not response.get_body_success() + ): + raise pytest.UsageError( + "Unable to resolve Cloud POS terminal group id: " + "GET /json/terminals did not return a successful response", + ) + + listing = response.get_body_data() + if not isinstance(listing, list) or not listing: + break + + for terminal in listing: + listed_terminal_id = terminal.get("id") + terminal_code = terminal.get("code") + if terminal_id not in {listed_terminal_id, terminal_code}: + continue + + group_id = terminal.get("group_id") + if group_id is None: + raise pytest.UsageError( + f"Unable to resolve Cloud POS terminal group id: " + f"terminal {terminal_id} has no group_id", + ) + return str(group_id) + + if len(listing) < limit: + break + + raise pytest.UsageError( + f"Unable to resolve Cloud POS terminal group id for terminal {terminal_id}", + ) + + +@pytest.fixture(scope="session") +def cloud_pos_terminal_id() -> str: + """Return terminal id used by SSE E2E tests.""" + return _require_first_env(*TERMINAL_ID_ENVS) + + +@pytest.fixture(scope="session") +def cloud_pos_base_url() -> str: + """Return dev-backed base URL used by SSE E2E tests.""" + return _validate_base_url( + _require_first_env(*BASE_URL_ENVS), + *BASE_URL_ENVS, + ) + + +@pytest.fixture(scope="session") +def cloud_pos_terminal_group_id( + cloud_pos_base_url: str, + cloud_pos_terminal_id: str, +) -> str: + """Return terminal group id from env or resolve it via /json/terminals.""" + explicit_group_id = _get_first_env(*TERMINAL_GROUP_ID_ENVS) + if explicit_group_id: + return explicit_group_id + + return _resolve_terminal_group_id( + base_url=cloud_pos_base_url, + default_api_key=_require_first_env(*DEFAULT_API_KEY_ENVS), + partner_api_key=_require_first_env(*PARTNER_API_KEY_ENVS), + terminal_id=cloud_pos_terminal_id, + ) + + +@pytest.fixture(scope="session") +def cloud_pos_events_sdk( + cloud_pos_base_url: str, + cloud_pos_terminal_group_id: str, +) -> Sdk: + """Return SDK configured for Cloud POS creation plus SSE subscription.""" + credential_resolver = ScopedCredentialResolver( + default_api_key=_require_first_env(*DEFAULT_API_KEY_ENVS), + terminal_group_api_keys={ + cloud_pos_terminal_group_id: _require_first_env( + *TERMINAL_GROUP_API_KEY_ENVS, + ), + }, + ) + sdk = Sdk( + is_production=False, + credential_resolver=credential_resolver, + ) + sdk.get_client().url = cloud_pos_base_url + return sdk + + +def pytest_collection_modifyitems( + config: pytest.Config, # noqa: ARG001 + items: list[pytest.Item], +) -> None: + """Skip SSE E2E tests when the required credentials are missing.""" + if _has_event_manager_e2e_env(): + return + + skip = pytest.mark.skip( + reason=( + "SSE E2E tests require E2E_API_KEY or API_KEY, " + "E2E_TERMINAL_GROUP_API_KEY_GROUP_DEFAULT or " + "TERMINAL_GROUP_API_KEY_GROUP_DEFAULT, " + "E2E_CLOUD_POS_TERMINAL_ID or CLOUD_POS_TERMINAL_ID, " + "E2E_NO_SANDBOX_BASE_URL or MSP_SDK_CUSTOM_BASE_URL, " + "and either CLOUD_POS_TERMINAL_GROUP_ID or PARTNER_API_KEY" + ), + ) + for item in items: + if item.nodeid.startswith(EVENT_MANAGER_E2E_NODE_PREFIXES): + item.add_marker(skip) diff --git a/tests/multisafepay/e2e/examples/event_manager/test_subscribe_events.py b/tests/multisafepay/e2e/examples/event_manager/test_subscribe_events.py new file mode 100644 index 0000000..c7d137b --- /dev/null +++ b/tests/multisafepay/e2e/examples/event_manager/test_subscribe_events.py @@ -0,0 +1,88 @@ +# 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/event_manager/subscribe_events.py.""" + +import time + +import pytest + +from multisafepay.api.base.response.custom_api_response import ( + CustomApiResponse, +) +from multisafepay.api.paths.events.event_manager import EventManager +from multisafepay.api.paths.events.stream import EventStream +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.sdk import Sdk + + +@pytest.fixture(scope="module") +def order_manager(cloud_pos_events_sdk: Sdk) -> OrderManager: + """Return OrderManager for Cloud POS event-stream tests.""" + return cloud_pos_events_sdk.get_order_manager() + + +@pytest.fixture(scope="module") +def event_manager(cloud_pos_events_sdk: Sdk) -> EventManager: + """Return EventManager for Cloud POS event-stream tests.""" + return cloud_pos_events_sdk.get_event_manager() + + +def _build_cloud_pos_order_request( + order_id: str, + terminal_id: str, +) -> OrderRequest: + """Create the Cloud POS order request used by the SSE example.""" + return ( + OrderRequest() + .add_type("redirect") + .add_order_id(order_id) + .add_description("Cloud POS order for event streaming") + .add_amount(100) + .add_currency("EUR") + .add_gateway_info( + { + "terminal_id": terminal_id, + }, + ) + ) + + +def test_subscribe_order_events_opens_stream( + order_manager: OrderManager, + event_manager: EventManager, + cloud_pos_terminal_group_id: str, + cloud_pos_terminal_id: str, +) -> None: + """Create a Cloud POS order and verify the SSE stream opens successfully.""" + order_id = f"cloud-pos-events-e2e-{int(time.time())}" + + create_response = order_manager.create( + request_order=_build_cloud_pos_order_request( + order_id=order_id, + terminal_id=cloud_pos_terminal_id, + ), + terminal_group_id=cloud_pos_terminal_group_id, + ) + + assert isinstance(create_response, CustomApiResponse) + assert create_response.get_status_code() == 200 + assert create_response.get_body_success() is True + + order = create_response.get_data() + assert isinstance(order, Order) + assert order.order_id == order_id + assert order.events_token or order.event_token + assert order.events_stream_url or order.event_stream_url + + with event_manager.subscribe_order_events(order, timeout=10.0) as stream: + assert isinstance(stream, EventStream) + assert stream.closed is False + + assert stream.closed is True 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/events/stream/test_unit_event_stream.py b/tests/multisafepay/unit/api/path/events/stream/test_unit_event_stream.py new file mode 100644 index 0000000..eb09647 --- /dev/null +++ b/tests/multisafepay/unit/api/path/events/stream/test_unit_event_stream.py @@ -0,0 +1,186 @@ +# 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 event-path stream contracts.""" + +from __future__ import annotations + +import io + +import pytest + +from multisafepay.api.paths.events.stream import Event, EventData, EventStream +from multisafepay.client.sse import ServerSentEventStream + +EVENTS_STREAM_URL = "https://testapi.multisafepay.com/events/stream/" +EVENTS_TOKEN = "events-token" +LAST_EVENT_ID = "last-10" +PING_PAYLOAD = b"data: ping\n\n" + + +class _FakeStreamingResponse: + """Small streaming response stub used for unit testing.""" + + def __init__(self: _FakeStreamingResponse, payload: bytes) -> None: + self._buffer = io.BytesIO(payload) + self.closed = False + + def readline(self: _FakeStreamingResponse) -> bytes: + return self._buffer.readline() + + def close(self: _FakeStreamingResponse) -> None: + self._buffer.close() + self.closed = True + + +def test_open_builds_expected_headers(monkeypatch: pytest.MonkeyPatch) -> None: + """Build event-specific auth headers before opening the generic SSE stream.""" + captured: dict[str, object] = {} + + def fake_open( + url: str, + headers: dict[str, str] | None = None, + timeout: float = 30.0, + ) -> ServerSentEventStream: + captured["url"] = url + captured["headers"] = headers + captured["timeout"] = timeout + return ServerSentEventStream( + response=_FakeStreamingResponse(PING_PAYLOAD), + ) + + monkeypatch.setattr( + "multisafepay.api.paths.events.stream.ServerSentEventStream.open", + fake_open, + ) + + stream = EventStream.open( + events_token=EVENTS_TOKEN, + events_stream_url=EVENTS_STREAM_URL, + last_event_id=LAST_EVENT_ID, + timeout=9.5, + ) + event = next(stream) + + assert isinstance(event, Event) + assert event.data == "ping" + assert captured["url"] == EVENTS_STREAM_URL + assert captured["timeout"] == 9.5 + headers = captured["headers"] + assert headers["Authorization"] == f"Bearer {EVENTS_TOKEN}" + assert headers["Accept"] == "text/event-stream" + assert headers["Cache-Control"] == "no-cache" + assert headers["Last-Event-ID"] == LAST_EVENT_ID + + +def test_wraps_generic_sse_messages_as_event_contracts() -> None: + """Adapt generic SSE messages into the events-path Event contract.""" + payload = ( + b"event: order.updated\n" + b"id: 15\n" + b"retry: 1000\n" + b'data: {"status": "completed", "order_id": "123"}\n\n' + ) + stream = EventStream(response=_FakeStreamingResponse(payload)) + + event = next(stream) + + assert isinstance(event, Event) + assert event.event == "order.updated" + assert event.event_id == "15" + assert event.retry == 1000 + assert isinstance(event.data, EventData) + assert event.data.status == "completed" + assert event.data.order_id == "123" + + +def test_event_from_dict_adapts_nested_payloads() -> None: + """Build nested event payload models through the common from_dict path.""" + event = Event.from_dict( + { + "event": "order.updated", + "data": { + "status": "processing", + "data": { + "status": "completed", + "order_id": "nested-1", + }, + }, + }, + ) + + assert event is not None + assert isinstance(event.data, EventData) + assert event.data.status == "processing" + assert isinstance(event.data.data, EventData) + assert event.data.data.status == "completed" + assert event.data.data.order_id == "nested-1" + + +def test_event_data_from_dict_adapts_nested_list_payloads() -> None: + """Build nested EventData items when payload data contains a list.""" + payload = EventData.from_dict( + { + "status": "processing", + "data": [ + { + "status": "completed", + "order_id": "nested-2", + }, + "keep-me", + ], + }, + ) + + assert payload is not None + assert payload.status == "processing" + assert isinstance(payload.data, list) + assert isinstance(payload.data[0], EventData) + assert payload.data[0].status == "completed" + assert payload.data[0].order_id == "nested-2" + assert payload.data[1] == "keep-me" + + +def test_event_from_dict_adapts_list_payload_items() -> None: + """Build top-level event payload lists through the common from_dict path.""" + event = Event.from_dict( + { + "event": "order.batch", + "data": [ + { + "status": "completed", + "order_id": "batch-1", + }, + "tail", + ], + }, + ) + + assert event is not None + assert isinstance(event.data, list) + assert isinstance(event.data[0], EventData) + assert event.data[0].status == "completed" + assert event.data[0].order_id == "batch-1" + assert event.data[1] == "tail" + + +def test_from_dict_returns_none_for_none_payload() -> None: + """Return None when from_dict receives None input.""" + assert Event.from_dict(None) is None + assert EventData.from_dict(None) is None + + +def test_closes_wrapped_stream_on_eof() -> None: + """Close the wrapped generic stream when EOF is reached.""" + response = _FakeStreamingResponse(payload=b"") + stream = EventStream(response=response) + + with pytest.raises(StopIteration): + next(stream) + + assert response.closed is True + assert stream.closed is True diff --git a/tests/multisafepay/unit/api/path/events/test_unit_event_manager.py b/tests/multisafepay/unit/api/path/events/test_unit_event_manager.py new file mode 100644 index 0000000..efbe3d0 --- /dev/null +++ b/tests/multisafepay/unit/api/path/events/test_unit_event_manager.py @@ -0,0 +1,121 @@ +# 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 event manager subscription helpers.""" + +from typing import Optional +from unittest.mock import MagicMock + +import pytest + +from multisafepay.api.paths.events.event_manager import EventManager +from multisafepay.api.paths.orders.response.order_response import Order + +TEST_EVENTS_STREAM_URL = "https://testapi.multisafepay.com/events/stream/" +ORDER_EVENTS_STREAM_URL = "https://stream.example/events/stream/" +LEGACY_EVENTS_STREAM_URL = "https://legacy.example/events/stream/" +MISSING_EVENTS_ERROR = "events_token/events_stream_url" + + +def _patch_event_stream_open( + monkeypatch: pytest.MonkeyPatch, +) -> tuple[dict[str, object], object]: + """Patch EventStream.open and return capture dict plus sentinel stream.""" + captured: dict[str, object] = {} + expected_stream = object() + + def fake_open( + events_token: str, + events_stream_url: str, + last_event_id: Optional[str] = None, + timeout: float = 30.0, + ) -> object: + captured["events_token"] = events_token + captured["events_stream_url"] = events_stream_url + captured["last_event_id"] = last_event_id + captured["timeout"] = timeout + return expected_stream + + monkeypatch.setattr( + "multisafepay.api.paths.events.event_manager.EventStream.open", + fake_open, + ) + + return captured, expected_stream + + +def test_subscribe_events_delegates_to_stream_open( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Delegate direct subscriptions to EventStream.open.""" + captured, expected_stream = _patch_event_stream_open(monkeypatch) + + manager = EventManager(MagicMock()) + stream = manager.subscribe_events( + events_token="token-abc", + events_stream_url=TEST_EVENTS_STREAM_URL, + last_event_id="last-15", + timeout=10.0, + ) + + assert stream is expected_stream + assert captured["events_token"] == "token-abc" + assert captured["events_stream_url"] == TEST_EVENTS_STREAM_URL + assert captured["last_event_id"] == "last-15" + assert captured["timeout"] == 10.0 + + +def test_subscribe_order_events_uses_plural_fields( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Read events credentials from events_* fields when present.""" + captured, expected_stream = _patch_event_stream_open(monkeypatch) + + manager = EventManager(MagicMock()) + order = Order( + order_id="order-1", + events_token="events-token", + events_stream_url=ORDER_EVENTS_STREAM_URL, + ) + + stream = manager.subscribe_order_events(order) + + assert stream is expected_stream + assert captured["events_token"] == "events-token" + assert captured["events_stream_url"] == ORDER_EVENTS_STREAM_URL + + +def test_subscribe_order_events_falls_back_to_legacy_fields( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Support old event_* field names for backward compatibility.""" + captured, expected_stream = _patch_event_stream_open(monkeypatch) + + manager = EventManager(MagicMock()) + order = Order( + order_id="order-2", + event_token="legacy-token", + event_stream_url=LEGACY_EVENTS_STREAM_URL, + ) + + stream = manager.subscribe_order_events(order) + + assert stream is expected_stream + assert captured["events_token"] == "legacy-token" + assert captured["events_stream_url"] == LEGACY_EVENTS_STREAM_URL + + +def test_subscribe_order_events_requires_token_and_stream_url() -> None: + """Raise a clear error when event credentials are missing in order.""" + manager = EventManager(MagicMock()) + order = Order(order_id="order-3") + + with pytest.raises( + ValueError, + match=MISSING_EVENTS_ERROR, + ): + manager.subscribe_order_events(order) 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/orders/response/test_unit_order_response.py b/tests/multisafepay/unit/api/path/orders/response/test_unit_order_response.py new file mode 100644 index 0000000..60a5b5c --- /dev/null +++ b/tests/multisafepay/unit/api/path/orders/response/test_unit_order_response.py @@ -0,0 +1,74 @@ +# 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 order response event fields compatibility.""" + +from typing import Optional + +from multisafepay.api.paths.orders.response.order_response import Order + +PLURAL_EVENTS_TOKEN = "token-123" +PLURAL_EVENTS_URL = "wss://testapi.multisafepay.com/events/" +PLURAL_EVENTS_STREAM_URL = "https://testapi.multisafepay.com/events/stream/" +LEGACY_EVENTS_TOKEN = "legacy-token" +LEGACY_EVENTS_URL = "wss://legacy.example.com/events/" +LEGACY_EVENTS_STREAM_URL = "https://legacy.example.com/events/stream/" + + +def _assert_event_fields( + order: Optional[Order], + expected_token: str, + expected_url: str, + expected_stream_url: str, +) -> None: + """Assert both plural and legacy event fields are populated consistently.""" + assert order is not None + + assert order.events_token == expected_token + assert order.events_url == expected_url + assert order.events_stream_url == expected_stream_url + assert order.event_token == expected_token + assert order.event_url == expected_url + assert order.event_stream_url == expected_stream_url + + +def test_from_dict_maps_plural_event_fields_to_legacy_aliases() -> None: + """Map events_* fields to both plural and legacy singular attributes.""" + data = { + "order_id": "order-1", + "events_token": PLURAL_EVENTS_TOKEN, + "events_url": PLURAL_EVENTS_URL, + "events_stream_url": PLURAL_EVENTS_STREAM_URL, + } + + order = Order.from_dict(data) + + _assert_event_fields( + order=order, + expected_token=PLURAL_EVENTS_TOKEN, + expected_url=PLURAL_EVENTS_URL, + expected_stream_url=PLURAL_EVENTS_STREAM_URL, + ) + + +def test_from_dict_maps_legacy_event_fields_to_plural_names() -> None: + """Map event_* fields to newer plural names for consistency.""" + data = { + "order_id": "order-2", + "event_token": LEGACY_EVENTS_TOKEN, + "event_url": LEGACY_EVENTS_URL, + "event_stream_url": LEGACY_EVENTS_STREAM_URL, + } + + order = Order.from_dict(data) + + _assert_event_fields( + order=order, + expected_token=LEGACY_EVENTS_TOKEN, + expected_url=LEGACY_EVENTS_URL, + expected_stream_url=LEGACY_EVENTS_STREAM_URL, + ) 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..a7fa84c 100644 --- a/tests/multisafepay/unit/test_unit_sdk.py +++ b/tests/multisafepay/unit/test_unit_sdk.py @@ -7,10 +7,45 @@ """Unit tests for SDK-level environment/base URL guardrails.""" +from unittest.mock import MagicMock + import pytest from multisafepay import Sdk +from multisafepay.api.paths.events.event_manager import EventManager 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 +99,55 @@ 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_returns_event_manager() -> None: + """Expose EventManager through SDK convenience getter.""" + sdk = Sdk( + api_key="mock_api_key", + is_production=False, + transport=MagicMock(), + ) + + assert isinstance(sdk.get_event_manager(), EventManager) + + +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}"