diff --git a/README.md b/README.md index b38faba..da768b4 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ The Python SDK offers a clean, type-safe API following Python best practices whi - **Secret Resolver** - **Telemetry & Observability** - **Data Anonymization Service** +- **Print Service** ## Requirements and Setup @@ -77,6 +78,7 @@ Each module has comprehensive usage guides: - [ObjectStore](src/sap_cloud_sdk/objectstore/user-guide.md) - [Secret Resolver](src/sap_cloud_sdk/core/secret_resolver/user-guide.md) - [Telemetry](src/sap_cloud_sdk/core/telemetry/user-guide.md) +- [Print](src/sap_cloud_sdk/print/user-guide.md) - [Data Anonymization](src/sap_cloud_sdk/core/data_anonymization/user-guide.md) ## Support, Feedback, Contributing diff --git a/pyproject.toml b/pyproject.toml index 401a762..b1eb3c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sap-cloud-sdk" -version = "0.23.2" +version = "0.24.1" description = "SAP Cloud SDK for Python" readme = "README.md" license = "Apache-2.0" diff --git a/src/sap_cloud_sdk/core/telemetry/module.py b/src/sap_cloud_sdk/core/telemetry/module.py index ef67a34..8901796 100644 --- a/src/sap_cloud_sdk/core/telemetry/module.py +++ b/src/sap_cloud_sdk/core/telemetry/module.py @@ -16,6 +16,7 @@ class Module(str, Enum): DMS = "dms" EXTENSIBILITY = "extensibility" OBJECTSTORE = "objectstore" + PRINT = "print" TELEMETRY = "telemetry" def __str__(self) -> str: diff --git a/src/sap_cloud_sdk/core/telemetry/operation.py b/src/sap_cloud_sdk/core/telemetry/operation.py index 178997a..7dbb29b 100644 --- a/src/sap_cloud_sdk/core/telemetry/operation.py +++ b/src/sap_cloud_sdk/core/telemetry/operation.py @@ -81,6 +81,14 @@ class Operation(str, Enum): AICORE_SET_CONFIG = "set_aicore_config" AICORE_AUTO_INSTRUMENT = "auto_instrument" + # Print Operations + PRINT_LIST_QUEUES = "list_queues" + PRINT_CREATE_QUEUE = "create_queue" + PRINT_GET_PROFILES = "get_print_profiles" + PRINT_UPLOAD_DOCUMENT = "upload_document" + PRINT_CREATE_TASK = "create_print_task" + PRINT_CREATE_CLIENT = "create_client" + # DMS Operations DMS_ONBOARD_REPOSITORY = "onboard_repository" DMS_GET_REPOSITORY = "get_repository" diff --git a/src/sap_cloud_sdk/print/__init__.py b/src/sap_cloud_sdk/print/__init__.py new file mode 100644 index 0000000..6939709 --- /dev/null +++ b/src/sap_cloud_sdk/print/__init__.py @@ -0,0 +1,106 @@ +"""SAP Cloud SDK for Python - Print module + +The create_client() function loads credentials from mounts/env vars and +returns a configured PrintClient. + +Usage: + from sap_cloud_sdk.print import create_client, PrintQueue, PrintContent, PrintTask + + client = create_client() + + # List queues + queues = client.list_queues() + + # Upload a document and print it + with open("invoice.pdf", "rb") as f: + doc_id = client.upload_document(f, filename="invoice.pdf") + + task = PrintTask( + item_id=doc_id, + qname="my-queue", + print_contents=[PrintContent(object_key=doc_id, document_name="invoice.pdf")], + ) + client.create_print_task(task) +""" + +from __future__ import annotations + +from typing import Optional + +from sap_cloud_sdk.print._models import ( + PrintContent, + PrintProfile, + PrintQueue, + PrintTask, + PrintTaskMetadata, +) +from sap_cloud_sdk.print.config import load_from_env_or_mount, PrintConfig +from sap_cloud_sdk.print._http import PrintHttp, TokenProvider +from sap_cloud_sdk.print.client import PrintClient +from sap_cloud_sdk.print.exceptions import ( + PrintError, + ClientCreationError, + ConfigError, + HttpError, + PrintOperationError, +) + +from sap_cloud_sdk.core.telemetry import ( + Module, + Operation, + record_error_metric as _record_error_metric, +) + + +def create_client( + *, + instance: Optional[str] = None, + config: Optional[PrintConfig] = None, + _telemetry_source: Optional[Module] = None, +) -> PrintClient: + """Create a PrintClient with secret resolution and OAuth setup. + + Args: + instance: Instance name used for secret resolution. Defaults to "default". + config: Optional explicit PrintConfig, bypasses secret resolution. + _telemetry_source: Internal parameter for telemetry. Not for external use. + + Returns: + Configured PrintClient. + + Raises: + ClientCreationError: If client creation fails. + """ + try: + binding = config or load_from_env_or_mount(instance) + tp = TokenProvider(binding) + http = PrintHttp(config=binding, token_provider=tp) + return PrintClient(http, _telemetry_source=_telemetry_source) + except Exception as e: + _record_error_metric( + Module.PRINT, + _telemetry_source, + Operation.PRINT_CREATE_CLIENT, + ) + raise ClientCreationError(f"failed to create print client: {e}") from e + + +__all__ = [ + # Models + "PrintQueue", + "PrintProfile", + "PrintContent", + "PrintTask", + "PrintTaskMetadata", + "PrintConfig", + # Factory + "create_client", + # Client + "PrintClient", + # Exceptions + "PrintError", + "ClientCreationError", + "ConfigError", + "HttpError", + "PrintOperationError", +] diff --git a/src/sap_cloud_sdk/print/_http.py b/src/sap_cloud_sdk/print/_http.py new file mode 100644 index 0000000..ce12da7 --- /dev/null +++ b/src/sap_cloud_sdk/print/_http.py @@ -0,0 +1,186 @@ +"""HTTP transport and OAuth utilities for SAP Print Service.""" + +from __future__ import annotations + +import base64 +import json +import logging +from typing import Any, Dict, Optional, Protocol + +import requests +from requests import Response +from requests.exceptions import RequestException +from oauthlib.oauth2 import BackendApplicationClient +from requests_oauthlib import OAuth2Session + +from sap_cloud_sdk.print.config import PrintConfig +from sap_cloud_sdk.print.exceptions import HttpError + +logger = logging.getLogger(__name__) + + +class AbstractTokenProvider(Protocol): + """Protocol for token providers — allows injection of mock providers in tests.""" + + def get_token(self) -> str: ... + def resolve_username(self) -> str: ... + + +class TokenProvider: + """Provides OAuth2 access tokens via client credentials flow.""" + + def __init__(self, config: PrintConfig) -> None: + self._config = config + client = BackendApplicationClient(client_id=config.client_id) + self._session = OAuth2Session(client=client) + self._cached_token: Optional[str] = None + + def get_token(self) -> str: + """Return a valid bearer token for the Print Service. + + Returns: + A non-empty OAuth2 access token string. + + Raises: + HttpError: If the token response is missing an access_token or + token acquisition fails. + """ + + try: + token: Dict[str, Any] = self._session.fetch_token( + token_url=self._config.token_url, + client_id=self._config.client_id, + client_secret=self._config.client_secret, + include_client_id=True, + ) + except Exception as e: + logger.error("failed to acquire token: %s", e) + raise HttpError(f"failed to acquire token: {e}") from e + access_token = token.get("access_token") + if not access_token: + raise HttpError("token response missing access_token") + self._cached_token = str(access_token) + return self._cached_token + + def resolve_username(self) -> str: + """Resolve a username from the current access token claims. + + Returns the ``user_name`` JWT claim when present (interactive user + flows), otherwise falls back to ``client_id`` (client-credentials / + technical-user flows). + """ + token = self._cached_token or self.get_token() + try: + payload_b64 = token.split(".")[1] + # JWT base64 uses URL-safe alphabet without padding + padding = 4 - len(payload_b64) % 4 + if padding != 4: + payload_b64 += "=" * padding + claims = json.loads(base64.urlsafe_b64decode(payload_b64)) + return str( + claims.get("user_name") + or claims.get("client_id") + or self._config.client_id + ) + except Exception: + logger.debug("could not decode JWT claims, falling back to client_id") + return self._config.client_id + + +class PrintHttp: + """HTTP client for SAP Print Service.""" + + def __init__( + self, + config: PrintConfig, + token_provider: AbstractTokenProvider, + session: Optional[requests.Session] = None, + ) -> None: + self._config = config + self._token_provider = token_provider + self._session = session or requests.Session() + self._base_url = config.url.rstrip("/") + + def get_username(self) -> str: + """Resolve the username from the current OAuth token (or fall back to client_id).""" + return self._token_provider.resolve_username() + + def _auth_headers(self) -> Dict[str, str]: + token = self._token_provider.get_token() + return {"Authorization": f"Bearer {token}"} + + def _request( + self, + method: str, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + json: Optional[Any] = None, + data: Optional[Any] = None, + files: Optional[Any] = None, + extra_headers: Optional[Dict[str, str]] = None, + ) -> Response: + url = f"{self._base_url}/{path.lstrip('/')}" + headers = self._auth_headers() + if extra_headers: + headers.update(extra_headers) + + try: + resp = self._session.request( + method=method, + url=url, + headers=headers, + params=params, + json=json, + data=data, + files=files, + ) + except RequestException as e: + logger.error("request failed [%s %s]: %s", method, url, e) + raise HttpError(f"request failed: {e}") from e + + if 200 <= resp.status_code < 300: + return resp + + text: str = "" + try: + text = resp.text + except Exception: + text = "" + + raise HttpError( + f"HTTP {resp.status_code} for {method} {url}", + status_code=resp.status_code, + response_text=text, + ) + + def get( + self, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Response: + return self._request("GET", path, params=params, extra_headers=headers) + + def put( + self, + path: str, + *, + json: Optional[Any] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Response: + return self._request("PUT", path, json=json, extra_headers=headers) + + def post( + self, + path: str, + *, + json: Optional[Any] = None, + data: Optional[Any] = None, + files: Optional[Any] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Response: + return self._request( + "POST", path, json=json, data=data, files=files, extra_headers=headers + ) diff --git a/src/sap_cloud_sdk/print/_models.py b/src/sap_cloud_sdk/print/_models.py new file mode 100644 index 0000000..c386c91 --- /dev/null +++ b/src/sap_cloud_sdk/print/_models.py @@ -0,0 +1,169 @@ +"""Data models for SAP Print Service.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class PrintQueue: + """Represents a print queue. + + Attributes: + qname: Queue name (A-Z, a-z, 0-9, underscore, hyphen; max 32 chars). + qdescription: Human-readable queue description. + qformat: Print format identifier (e.g., acrobat6.xdc). + qformat_descript: Format description (e.g., PDF). + cleanup_prd: Document retention period in days (1-7). + tech_user_name: Technical user assigned to the queue. + location_id: Physical location identifier. + location_id_type: Location type identifier. + creator: Email of the queue creator. + """ + + qname: str + qdescription: str = "" + qformat: str = "" + qformat_descript: str = "" + cleanup_prd: int = 1 + tech_user_name: str = "" + location_id: str = "" + location_id_type: str = "" + creator: str = "" + + def to_dict(self) -> dict: + return { + "qname": self.qname, + "qdescription": self.qdescription, + "qformat": self.qformat, + "qformatDescript": self.qformat_descript, + "cleanupPrd": self.cleanup_prd, + "techUserName": self.tech_user_name, + "locationId": self.location_id, + "locationIdType": self.location_id_type, + "creator": self.creator, + } + + @classmethod + def from_dict(cls, data: dict) -> "PrintQueue": + return cls( + qname=data.get("qname", ""), + qdescription=data.get("qdescription", ""), + qformat=data.get("qformat", ""), + qformat_descript=data.get("qformatDescript", ""), + cleanup_prd=data.get("cleanupPrd", 1), + tech_user_name=data.get("techUserName", ""), + location_id=data.get("locationId", ""), + location_id_type=data.get("locationIdType", ""), + creator=data.get("creator", ""), + ) + + +@dataclass +class PrintProfile: + """Represents a print profile for a queue. + + Attributes: + queue_name: Name of the associated print queue. + profile_name: Profile identifier used when creating print tasks. + profile_params: Reserved for future profile parameter details (currently empty). + profile_status: Profile status (e.g., OK). + """ + + queue_name: str + profile_name: str + profile_params: Optional[str] = None + profile_status: str = "" + + @classmethod + def from_dict(cls, data: dict) -> "PrintProfile": + return cls( + queue_name=data.get("queueName", ""), + profile_name=data.get("profileName", ""), + profile_params=data.get("profileParams"), + profile_status=data.get("profileStatus", ""), + ) + + +@dataclass +class PrintContent: + """A single document included in a print task. + + Attributes: + object_key: Document ID returned by upload_document(). + document_name: Display name for the document. Attachments must include + the file extension (e.g., attachment.pdf). + """ + + object_key: str + document_name: str + + def to_dict(self) -> dict: + return { + "objectKey": self.object_key, + "documentName": self.document_name, + } + + +@dataclass +class PrintTaskMetadata: + """Optional metadata to attach to a print task. + + Attributes: + version: Metadata schema version (required when metadata is provided). + business_user: Business user identifier. + object_node_type: Object node type identifier. + """ + + version: float + business_user: str = "" + object_node_type: str = "" + + def to_dict(self) -> dict: + return { + "version": self.version, + "business_metadata": { + "business_user": self.business_user, + "object_node_type": self.object_node_type, + }, + } + + +@dataclass +class PrintTask: + """Describes a print task to be sent to a queue. + + The first item in print_contents whose object_key matches item_id is treated + as the main document; all others are attachments. + + Attributes: + item_id: The object_key of the main document (must match one entry in print_contents). + qname: Target print queue name. + print_contents: List of documents (main + attachments). + number_of_copies: Number of copies to print. + username: Name of the user initiating the print. + profile_name: Optional profile name from get_print_profiles(). + metadata: Optional structured metadata. + """ + + item_id: str + qname: str + print_contents: list[PrintContent] + number_of_copies: int = 1 + username: str = "" + profile_name: Optional[str] = None + metadata: Optional[PrintTaskMetadata] = None + + def to_body(self) -> dict: + body: dict = { + "numberOfCopies": self.number_of_copies, + "username": self.username, + "qname": self.qname, + "printContents": [c.to_dict() for c in self.print_contents], + } + if self.profile_name is not None: + body["profileName"] = self.profile_name + if self.metadata is not None: + body["metadata"] = self.metadata.to_dict() + return body diff --git a/src/sap_cloud_sdk/print/client.py b/src/sap_cloud_sdk/print/client.py new file mode 100644 index 0000000..b062f05 --- /dev/null +++ b/src/sap_cloud_sdk/print/client.py @@ -0,0 +1,198 @@ +"""SAP Print Service client implementation.""" + +from __future__ import annotations + +import logging +from typing import IO, Optional, Union + +from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics +from sap_cloud_sdk.print._http import PrintHttp +from sap_cloud_sdk.print.exceptions import HttpError, PrintOperationError +from sap_cloud_sdk.print._models import PrintProfile, PrintQueue, PrintTask + +_QUEUES_PATH = "qm/api/v1/rest/queues" +_DOCUMENTS_PATH = "dm/api/v1/rest/print-documents" +_TASKS_PATH = "qm/api/v1/rest/print-tasks" + +_IF_NONE_MATCH = {"If-None-Match": "*"} + +logger = logging.getLogger(__name__) + + +class PrintClient: + """Client for SAP Print Service operations. + + Note: + Do not instantiate PrintClient directly. Use create_client() from + sap_cloud_sdk.print instead, which handles environment detection, + secret resolution and OAuth setup. + + Example: + ```python + from sap_cloud_sdk.print import create_client, PrintQueue, PrintContent, PrintTask + + client = create_client() + + # List available print queues + queues = client.list_queues() + + # Upload a document + with open("invoice.pdf", "rb") as f: + document_id = client.upload_document(f) + + # Create a print task + task = PrintTask( + item_id=document_id, + qname="my-queue", + print_contents=[PrintContent(object_key=document_id, document_name="invoice.pdf")], + ) + client.create_print_task(task) + ``` + """ + + def __init__( + self, http: PrintHttp, _telemetry_source: Optional[Module] = None + ) -> None: + self._http = http + self._telemetry_source = _telemetry_source + + @record_metrics(Module.PRINT, Operation.PRINT_LIST_QUEUES) + def list_queues(self) -> list[PrintQueue]: + """Retrieve all print queues available in the tenant. + + Returns: + List of PrintQueue objects. + + Raises: + PrintOperationError: If the request fails or the response cannot be parsed. + """ + try: + resp = self._http.get(_QUEUES_PATH) + data = resp.json() + return [PrintQueue.from_dict(item) for item in data] + except HttpError as e: + logger.error("failed to list queues: %s", e) + raise PrintOperationError(f"failed to list queues: {e}") from e + except Exception as e: + logger.error("failed to parse list queues response: %s", e) + raise PrintOperationError( + f"failed to parse list queues response: {e}" + ) from e + + @record_metrics(Module.PRINT, Operation.PRINT_CREATE_QUEUE) + def create_queue(self, queue: PrintQueue) -> None: + """Create a print queue. + + Args: + queue: PrintQueue to create. The queue name in the body must match + the path parameter — this is enforced automatically. + + Raises: + PrintOperationError: If the request fails. + """ + try: + self._http.put( + f"{_QUEUES_PATH}/{queue.qname}", + json=queue.to_dict(), + headers=_IF_NONE_MATCH, + ) + except HttpError as e: + logger.error("failed to create queue '%s': %s", queue.qname, e) + raise PrintOperationError( + f"failed to create queue '{queue.qname}': {e}" + ) from e + + @record_metrics(Module.PRINT, Operation.PRINT_GET_PROFILES) + def get_print_profiles(self, qname: str) -> list[PrintProfile]: + """Fetch print profiles for a queue. + + Use the returned profile names when creating print tasks to send + profile parameters to the physical printer. + + Args: + qname: Name of the existing print queue. + + Returns: + List of PrintProfile objects for the queue. + + Raises: + PrintOperationError: If the request fails or the response cannot be parsed. + """ + try: + resp = self._http.get(f"{_QUEUES_PATH}/{qname}/profiles") + data = resp.json() + return [PrintProfile.from_dict(item) for item in data] + except HttpError as e: + logger.error("failed to get profiles for queue '%s': %s", qname, e) + raise PrintOperationError( + f"failed to get profiles for queue '{qname}': {e}" + ) from e + except Exception as e: + logger.error("failed to parse get profiles response: %s", e) + raise PrintOperationError( + f"failed to parse get profiles response: {e}" + ) from e + + @record_metrics(Module.PRINT, Operation.PRINT_UPLOAD_DOCUMENT) + def upload_document( + self, + file: Union[IO[bytes], bytes], + filename: str = "document", + scan: bool = True, + ) -> str: + """Upload a document to Print Service cloud storage. + + The returned document ID is used as the object_key in PrintContent + and as the item_id in PrintTask. + + Args: + file: File-like object (opened in binary mode) or raw bytes. + filename: Name for the uploaded file. + scan: Whether to enable virus scanning. Defaults to True. + + Returns: + Document ID (UUID string) to reference in create_print_task(). + + Raises: + PrintOperationError: If the upload fails. + """ + try: + headers = {**_IF_NONE_MATCH, "scan": str(scan).lower()} + resp = self._http.post( + _DOCUMENTS_PATH, + files={"file": (filename, file)}, + headers=headers, + ) + return resp.text.strip() + except HttpError as e: + logger.error("failed to upload document: %s", e) + raise PrintOperationError(f"failed to upload document: {e}") from e + + @record_metrics(Module.PRINT, Operation.PRINT_CREATE_TASK) + def create_print_task(self, task: PrintTask) -> None: + """Send a document to a print queue. + + The task.item_id must match the object_key of the main document in + task.print_contents. All other entries in print_contents are treated + as attachments. + + If task.username is empty, it is resolved automatically from the + OAuth token (``user_name`` claim) or falls back to the client ID. + + Args: + task: PrintTask describing the print job. + + Raises: + PrintOperationError: If the request fails. + """ + if not task.username: + task.username = self._http.get_username() + try: + self._http.put( + f"{_TASKS_PATH}/{task.item_id}", + json=task.to_body(), + headers=_IF_NONE_MATCH, + ) + except HttpError as e: + logger.error("failed to create print task: %s", e) + raise PrintOperationError(f"failed to create print task: {e}") from e diff --git a/src/sap_cloud_sdk/print/config.py b/src/sap_cloud_sdk/print/config.py new file mode 100644 index 0000000..f3a5e99 --- /dev/null +++ b/src/sap_cloud_sdk/print/config.py @@ -0,0 +1,120 @@ +"""Configuration and secret resolution for SAP Print Service. + +Loads service binding secrets from a mounted volume with environment fallback, +then normalizes into a unified PrintConfig model. + +Mount path convention: + /etc/secrets/appfnd/print/{instance}/ +Keys: + - url (Print service base URL, e.g. https://api.eu10.print.services.sap) + - uaa (JSON string with clientid, clientsecret, url, identityzone) + +Env fallback convention: + CLOUD_SDK_CFG_PRINT_{INSTANCE}_{FIELD_KEY} + e.g., CLOUD_SDK_CFG_PRINT_DEFAULT_URL +""" + +from dataclasses import dataclass +from typing import Optional +import json +import logging + +from sap_cloud_sdk.core.secret_resolver.resolver import ( + read_from_mount_and_fallback_to_env_var, +) +from sap_cloud_sdk.print.exceptions import ConfigError + +logger = logging.getLogger(__name__) + + +@dataclass +class PrintConfig: + """Service binding for SAP Print Service. + + Args: + url: Print service base URL (e.g., https://api.eu10.print.services.sap) + token_url: OAuth2 token endpoint + client_id: OAuth2 client id + client_secret: OAuth2 client secret + """ + + url: str + token_url: str + client_id: str + client_secret: str + + +@dataclass +class _BindingData: + """Raw binding secrets read via the secret resolver.""" + + url: str = "" + uaa: str = "" # JSON string: {clientid, clientsecret, url, identityzone} + + def validate(self) -> None: + if not self.url: + raise ValueError("url is required") + if not self.uaa: + raise ValueError("uaa is required") + + def to_config(self) -> PrintConfig: + try: + uaa = json.loads(self.uaa) + except json.JSONDecodeError as e: + raise ValueError(f"uaa is not valid JSON: {e}") from e + + client_id = uaa.get("clientid", "") + client_secret = uaa.get("clientsecret", "") + auth_url = uaa.get("url", "") + + if not client_id: + raise ValueError("uaa.clientid is required") + if not client_secret: + raise ValueError("uaa.clientsecret is required") + if not auth_url: + raise ValueError("uaa.url is required") + + token_url = auth_url.rstrip("/") + "/oauth/token" + + return PrintConfig( + url=self.url, + token_url=token_url, + client_id=client_id, + client_secret=client_secret, + ) + + +def load_from_env_or_mount(instance: Optional[str] = None) -> PrintConfig: + """Load Print configuration from mount with env fallback and normalize. + + Args: + instance: Logical instance name; defaults to "default" if not provided. + + Returns: + PrintConfig + + Raises: + ConfigError: If loading or validation fails. + """ + inst = instance or "default" + binding = _BindingData() + + try: + read_from_mount_and_fallback_to_env_var( + base_volume_mount="/etc/secrets/appfnd", + base_var_name="CLOUD_SDK_CFG", + module="print", + instance=inst, + target=binding, + ) + + binding.validate() + return binding.to_config() + + except Exception as e: + logger.error( + "failed to load print configuration for instance='%s': %s", inst, e + ) + raise ConfigError( + f"failed to load print configuration for instance='{inst}': {e}" + ) from e diff --git a/src/sap_cloud_sdk/print/exceptions.py b/src/sap_cloud_sdk/print/exceptions.py new file mode 100644 index 0000000..372a086 --- /dev/null +++ b/src/sap_cloud_sdk/print/exceptions.py @@ -0,0 +1,45 @@ +"""Exception classes for the Print module.""" + + +class PrintError(Exception): + """Base exception for all Print module errors.""" + + pass + + +class ClientCreationError(PrintError): + """Raised when Print client creation fails.""" + + pass + + +class ConfigError(PrintError): + """Raised when configuration or secret resolution fails.""" + + pass + + +class HttpError(PrintError): + """Raised for HTTP-related errors from Print Service. + + Attributes: + status_code: HTTP status code returned by the service, if available. + message: Human-readable error message. + response_text: Raw response payload for diagnostics, if available. + """ + + def __init__( + self, + message: str, + status_code: int | None = None, + response_text: str | None = None, + ): + super().__init__(message) + self.status_code = status_code + self.response_text = response_text + + +class PrintOperationError(PrintError): + """Raised when a Print operation fails.""" + + pass diff --git a/src/sap_cloud_sdk/print/py.typed b/src/sap_cloud_sdk/print/py.typed new file mode 100644 index 0000000..5ef2441 --- /dev/null +++ b/src/sap_cloud_sdk/print/py.typed @@ -0,0 +1 @@ +# Marker file for PEP 561 to indicate the 'print' package is typed. diff --git a/src/sap_cloud_sdk/print/user-guide.md b/src/sap_cloud_sdk/print/user-guide.md new file mode 100644 index 0000000..9945625 --- /dev/null +++ b/src/sap_cloud_sdk/print/user-guide.md @@ -0,0 +1,303 @@ +# Print Service User Guide + +This module provides a Python SDK for interacting with the SAP Print Service. It supports managing print queues, uploading documents, and creating print tasks. + +## Installation + +```bash +# Using pip +pip install sap-cloud-sdk +``` + +See further information about installation in the [main documentation](/README.md#installation). + +## Import + +```python +from sap_cloud_sdk.print import create_client +from sap_cloud_sdk.print import ( + PrintQueue, PrintProfile, PrintContent, PrintTask, PrintTaskMetadata, +) +from sap_cloud_sdk.print.exceptions import ( + ClientCreationError, ConfigError, PrintOperationError, HttpError, +) +``` + +--- + +## Getting Started + +Use `create_client()` to get a client with automatic configuration detection: + +```python +from sap_cloud_sdk.print import create_client + +# Load credentials from mounted secrets or environment variables +client = create_client(instance="my-instance") +``` + +You can also provide credentials directly: + +```python +from sap_cloud_sdk.print import create_client +from sap_cloud_sdk.print.config import PrintConfig + +config = PrintConfig( + url="https://api.eu10.print.services.sap", + client_id="your-client-id", + client_secret="your-client-secret", + token_url="https://your-subdomain.authentication.eu10.hana.ondemand.com/oauth/token", +) + +client = create_client(config=config) +``` + +> **`instance` refers to the instance name defined in your Cloud descriptor.** +> +> This name determines which set of credentials or mounted secrets to resolve from the environment. + +--- + +## Queue Management + +### List All Queues + +```python +queues = client.list_queues() + +for queue in queues: + print(f"{queue.qname} — {queue.qdescription}") + print(f" Format: {queue.qformat} ({queue.qformat_descript})") + print(f" Location: {queue.location_id}") +``` + +### Create a Queue + +```python +from sap_cloud_sdk.print import PrintQueue + +queue = PrintQueue( + qname="my-queue", + qdescription="Main invoice printer", + qformat="acrobat6.xdc", + tech_user_name="tech_user", + cleanup_prd=1, +) + +client.create_queue(queue) +``` +> **Note:** `tech_user` should be created firstly, please check the [official help guide]('https://help.sap.com/docs/SCP_PRINT_SERVICE/7615de0949ce441d8bc5df7725a6bfc6/b497cdaba35946e9a871bc098e881b69.html'). + +The queue `qname` must be unique and contain only A–Z, a–z, 0–9, underscores, or hyphens (max 32 characters). + +### Get Print Profiles + +Returns the print profiles defined for a queue. Use `profile_name` values when creating print tasks to pass profile parameters directly to the physical printer. + +```python +profiles = client.get_print_profiles("my-queue") + +for profile in profiles: + print(f"{profile.profile_name} — status={profile.profile_status}") +``` + +--- + +## Document Upload + +### Upload a Document + +Uploads a document to Print Service cloud storage and returns a document ID (UUID). This ID is used as the `object_key` in `PrintContent` and as the `item_id` in `PrintTask`. + +```python +# From a file on disk +with open("invoice.pdf", "rb") as f: + doc_id = client.upload_document(f, filename="invoice.pdf") + +print(f"Uploaded document ID: {doc_id}") +``` + +```python +# From in-memory bytes +import io + +content = io.BytesIO(b"%PDF-1.4 ...") +doc_id = client.upload_document(content, filename="invoice.pdf") +``` + +Set `scan=False` to skip virus scanning: + +```python +doc_id = client.upload_document(f, filename="invoice.pdf", scan=False) +``` + +--- + +## Print Tasks + +### Create a Print Task + +Sends a print job to a queue. The `task.item_id` must match the `object_key` of one entry in `task.print_contents` — that entry becomes the main document. + +```python +from sap_cloud_sdk.print import PrintContent, PrintTask + +task = PrintTask( + item_id=doc_id, + qname="my-queue", + print_contents=[PrintContent(object_key=doc_id, document_name="invoice.pdf")], +) + +client.create_print_task(task) +``` + +### Print with Multiple Documents + +Additional entries in `print_contents` are treated as attachments and must have filenames that include the extension (e.g., `attachment.pdf`). + +```python +with open("main.pdf", "rb") as f: + main_id = client.upload_document(f, filename="main.pdf") + +with open("attachment.pdf", "rb") as f: + att_id = client.upload_document(f, filename="attachment.pdf") + +task = PrintTask( + item_id=main_id, + qname="my-queue", + number_of_copies=2, + username="user@example.com", + print_contents=[ + PrintContent(object_key=main_id, document_name="main.pdf"), + PrintContent(object_key=att_id, document_name="attachment.pdf"), + ], +) + +client.create_print_task(task) +``` + +### Using a Print Profile + +```python +profiles = client.get_print_profiles("my-queue") +profile_name = profiles[0].profile_name # e.g., "Defaults" + +task = PrintTask( + item_id=doc_id, + qname="my-queue", + print_contents=[PrintContent(object_key=doc_id, document_name="doc.pdf")], + profile_name=profile_name, +) + +client.create_print_task(task) +``` + +### Print Task with Metadata + +```python +from sap_cloud_sdk.print import PrintTaskMetadata + +task = PrintTask( + item_id=doc_id, + qname="my-queue", + print_contents=[PrintContent(object_key=doc_id, document_name="doc.pdf")], + metadata=PrintTaskMetadata( + version=1.0, + business_user="user@example.com", + object_node_type="INVOICE", + ), +) + +client.create_print_task(task) +``` + +--- + +## Error Handling + +The Print module provides specific exceptions for different error scenarios: + +```python +from sap_cloud_sdk.print.exceptions import ( + ClientCreationError, + ConfigError, + PrintOperationError, + HttpError, +) + +try: + client = create_client() + queues = client.list_queues() +except ClientCreationError as e: + print(f"Could not connect to Print Service: {e}") +except ConfigError as e: + print(f"Configuration error: {e}") +except PrintOperationError as e: + print(f"Operation failed: {e}") +except HttpError as e: + print(f"HTTP error ({e.status_code}): {e.response_text}") +``` + +### Exception Hierarchy + +| Exception | Description | +|---|---| +| `ClientCreationError` | Client creation fails (bad config, missing credentials) | +| `ConfigError` | Secret resolution or config parsing fails | +| `PrintOperationError` | An API operation fails (wraps `HttpError`) | +| `HttpError` | Raw HTTP error with `status_code` and `response_text` | + +--- + +## Models + +### Queue Models + +- **`PrintQueue`**: Print queue — `qname` (required), `qdescription`, `qformat`, `qformat_descript`, `cleanup_prd` (1–7 days), `tech_user_name`, `location_id`, `location_id_type`, `creator` +- **`PrintProfile`**: Queue print profile — `queue_name`, `profile_name`, `profile_params`, `profile_status` + +### Task Models + +- **`PrintContent`**: Document reference — `object_key` (document ID from `upload_document()`), `document_name` (attachments must include file extension) +- **`PrintTask`**: Print job — `item_id` (main document's `object_key`), `qname`, `print_contents`, `number_of_copies` (default 1), `username`, `profile_name`, `metadata` +- **`PrintTaskMetadata`**: Optional task metadata — `version` (required when provided), `business_user`, `object_node_type` + +--- + +## Configuration + +### Service Binding + +- **Mount path**: `$SERVICE_BINDING_ROOT/print/{instance}/` (defaults to `/etc/secrets/appfnd/print/{instance}/`) +- **Required Keys**: `url` (Print Service API base URL), `uaa` (JSON string with XSUAA credentials) +- **Env var fallback**: `CLOUD_SDK_CFG_PRINT_{INSTANCE}_{FIELD}` (uppercased, hyphens in instance replaced with `_`) + +> **Note:** `SERVICE_BINDING_ROOT` defaults to `/etc/secrets/appfnd` when not set. See the [Secret Resolver guide](../core/secret_resolver/user-guide.md) for details. + +#### Mounted Secrets (Kubernetes) + +``` +$SERVICE_BINDING_ROOT/print/{instance}/ +├── url +└── uaa +``` + +#### Environment Variables + +```bash +# Example for Print Service with instance name "default" +export CLOUD_SDK_CFG_PRINT_DEFAULT_URL="https://api.eu10.print.services.sap" +export CLOUD_SDK_CFG_PRINT_DEFAULT_UAA='{"clientid":"...","clientsecret":"...","url":"https://subdomain.authentication.eu10.hana.ondemand.com"}' +``` + +#### UAA JSON Schema + +The `uaa` key must contain a JSON string with the XSUAA credentials: + +```json +{ + "clientid": "sb-xxx!bxxx|print!bxxx", + "clientsecret": "xxx", + "url": "https://subdomain.authentication.region.hana.ondemand.com" +} +``` diff --git a/tests/core/unit/telemetry/test_module.py b/tests/core/unit/telemetry/test_module.py index 8eb2549..e923577 100644 --- a/tests/core/unit/telemetry/test_module.py +++ b/tests/core/unit/telemetry/test_module.py @@ -17,6 +17,7 @@ def test_module_values(self): assert Module.DESTINATION.value == "destination" assert Module.OBJECTSTORE.value == "objectstore" assert Module.DMS.value == "dms" + assert Module.PRINT.value == "print" def test_module_str_representation(self): """Test that Module enum converts to string correctly.""" @@ -27,6 +28,7 @@ def test_module_str_representation(self): assert str(Module.DESTINATION) == "destination" assert str(Module.OBJECTSTORE) == "objectstore" assert str(Module.DMS) == "dms" + assert str(Module.PRINT) == "print" def test_module_is_string_enum(self): """Test that Module enum inherits from str.""" @@ -53,7 +55,7 @@ def test_module_in_collection(self): def test_all_modules_present(self): """Test that all expected modules are present.""" all_modules = list(Module) - assert len(all_modules) == 11 + assert len(all_modules) == 12 assert Module.AICORE in all_modules assert Module.AUDITLOG in all_modules assert Module.AUDITLOG_NG in all_modules @@ -63,6 +65,7 @@ def test_all_modules_present(self): assert Module.OBJECTSTORE in all_modules assert Module.DMS in all_modules assert Module.AGENT_MEMORY in all_modules + assert Module.PRINT in all_modules def test_module_iteration(self): """Test iterating over Module enum.""" @@ -75,3 +78,4 @@ def test_module_iteration(self): assert "objectstore" in module_values assert "dms" in module_values assert "extensibility" in module_values + assert "print" in module_values diff --git a/tests/core/unit/telemetry/test_operation.py b/tests/core/unit/telemetry/test_operation.py index 2fde7c2..8c4ca78 100644 --- a/tests/core/unit/telemetry/test_operation.py +++ b/tests/core/unit/telemetry/test_operation.py @@ -150,6 +150,15 @@ def test_dms_operations(self): assert Operation.DMS_APPEND_CONTENT_STREAM.value == "cmis_append_content_stream" assert Operation.DMS_CMIS_QUERY.value == "cmis_query" + def test_print_operations(self): + """Test Print operation values.""" + assert Operation.PRINT_CREATE_CLIENT.value == "create_client" + assert Operation.PRINT_LIST_QUEUES.value == "list_queues" + assert Operation.PRINT_CREATE_QUEUE.value == "create_queue" + assert Operation.PRINT_GET_PROFILES.value == "get_print_profiles" + assert Operation.PRINT_UPLOAD_DOCUMENT.value == "upload_document" + assert Operation.PRINT_CREATE_TASK.value == "create_print_task" + def test_operation_str_representation(self): """Test that Operation enum converts to string correctly.""" assert str(Operation.AUDITLOG_LOG) == "log" @@ -196,10 +205,11 @@ def test_operation_iteration(self): assert any("FRAGMENT" in op.name for op in all_operations) assert any("OBJECTSTORE" in op.name for op in all_operations) assert any("AICORE" in op.name for op in all_operations) + assert any("PRINT" in op.name for op in all_operations) def test_operation_count(self): """Test that we have the expected number of operations.""" all_operations = list(Operation) # 3 auditlog + 11 destination + 10 certificate + 10 fragment + 8 objectstore - # + 2 extensibility + 2 aicore + 23 dms + 4 agentgateway + 13 agent_memory + 5 data anonymization = 89 - assert len(all_operations) == 91 + # + 2 extensibility + 2 aicore + 23 dms + 4 agentgateway + 13 agent_memory + 5 data anonymization + 6 print = 97 + assert len(all_operations) == 96 diff --git a/tests/print/__init__.py b/tests/print/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/print/unit/__init__.py b/tests/print/unit/__init__.py new file mode 100644 index 0000000..98c502f --- /dev/null +++ b/tests/print/unit/__init__.py @@ -0,0 +1 @@ +"""Unit tests for print module.""" diff --git a/tests/print/unit/test_client.py b/tests/print/unit/test_client.py new file mode 100644 index 0000000..fcea6f3 --- /dev/null +++ b/tests/print/unit/test_client.py @@ -0,0 +1,287 @@ +"""Unit tests for PrintClient.""" + +import pytest +from unittest.mock import MagicMock +from requests import Response + +from sap_cloud_sdk.print.client import PrintClient +from sap_cloud_sdk.print._models import PrintContent, PrintProfile, PrintQueue, PrintTask +from sap_cloud_sdk.print.exceptions import HttpError, PrintOperationError + + +def _mock_response(status_code: int, json_data=None, text: str = "") -> Response: + resp = MagicMock(spec=Response) + resp.status_code = status_code + resp.json.return_value = json_data or [] + resp.text = text + return resp + + +class TestListQueues: + + def test_returns_queue_list(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response( + 200, + json_data=[ + {"qname": "q1", "qdescription": "Queue 1", "cleanupPrd": 3}, + {"qname": "q2", "qdescription": "Queue 2", "cleanupPrd": 1}, + ], + ) + + client = PrintClient(mock_http) + queues = client.list_queues() + + assert len(queues) == 2 + assert all(isinstance(q, PrintQueue) for q in queues) + assert queues[0].qname == "q1" + assert queues[1].qname == "q2" + + def test_returns_empty_list(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response(200, json_data=[]) + + client = PrintClient(mock_http) + assert client.list_queues() == [] + + def test_http_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.get.side_effect = HttpError("server error", status_code=500) + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to list queues"): + client.list_queues() + + def test_parse_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response(200, json_data="not-a-list") + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to parse list queues response"): + client.list_queues() + + +class TestCreateQueue: + + def test_creates_queue_successfully(self): + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + + queue = PrintQueue(qname="my-queue", qdescription="Test", cleanup_prd=2) + client = PrintClient(mock_http) + client.create_queue(queue) + + args, kwargs = mock_http.put.call_args + assert "my-queue" in args[0] + assert kwargs["json"]["qname"] == "my-queue" + assert kwargs["headers"]["If-None-Match"] == "*" + + def test_http_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.put.side_effect = HttpError("conflict", status_code=412) + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to create queue 'bad-q'"): + client.create_queue(PrintQueue(qname="bad-q")) + + +class TestGetPrintProfiles: + + def test_returns_profile_list(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response( + 200, + json_data=[ + {"queueName": "q1", "profileName": "default", "profileStatus": "OK"}, + ], + ) + + client = PrintClient(mock_http) + profiles = client.get_print_profiles("q1") + + assert len(profiles) == 1 + assert isinstance(profiles[0], PrintProfile) + assert profiles[0].profile_name == "default" + assert profiles[0].profile_status == "OK" + + def test_correct_path_used(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response(200, json_data=[]) + + client = PrintClient(mock_http) + client.get_print_profiles("q1") + + args, _ = mock_http.get.call_args + assert "q1/profiles" in args[0] + + def test_http_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.get.side_effect = HttpError("not found", status_code=400) + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to get profiles"): + client.get_print_profiles("no-queue") + + def test_parse_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.get.return_value = _mock_response(200, json_data="not-a-list") + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to parse get profiles response"): + client.get_print_profiles("q1") + + +class TestUploadDocument: + + def test_returns_document_id(self): + mock_http = MagicMock() + doc_id = "4056bb6c-f544-41d7-87e1-ffe818573e6e" + mock_http.post.return_value = _mock_response(201, text=doc_id + "\n") + + client = PrintClient(mock_http) + result = client.upload_document(b"PDF content", filename="invoice.pdf") + + assert result == doc_id + + def test_scan_header_passed(self): + mock_http = MagicMock() + mock_http.post.return_value = _mock_response( + 201, text="some-id" + ) + + client = PrintClient(mock_http) + client.upload_document(b"data", scan=False) + + _, kwargs = mock_http.post.call_args + assert kwargs["headers"]["scan"] == "false" + + def test_http_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.post.side_effect = HttpError("too large", status_code=413) + + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to upload document"): + client.upload_document(b"data") + + +class TestCreatePrintTask: + + def test_creates_task_successfully(self): + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + + task = PrintTask( + item_id="doc-id-1", + qname="q1", + print_contents=[ + PrintContent(object_key="doc-id-1", document_name="main.pdf") + ], + number_of_copies=2, + username="user@example.com", + ) + client = PrintClient(mock_http) + client.create_print_task(task) + + args, kwargs = mock_http.put.call_args + assert "doc-id-1" in args[0] + body = kwargs["json"] + assert body["qname"] == "q1" + assert body["numberOfCopies"] == 2 + assert body["username"] == "user@example.com" + assert len(body["printContents"]) == 1 + assert kwargs["headers"]["If-None-Match"] == "*" + + def test_optional_profile_name_included(self): + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + + task = PrintTask( + item_id="doc-id", + qname="q1", + print_contents=[PrintContent(object_key="doc-id", document_name="f.pdf")], + profile_name="custom-profile", + ) + client = PrintClient(mock_http) + client.create_print_task(task) + + _, kwargs = mock_http.put.call_args + assert kwargs["json"]["profileName"] == "custom-profile" + + def test_username_auto_resolved_when_empty(self): + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + mock_http.get_username.return_value = "auto@example.com" + + task = PrintTask( + item_id="doc-id", + qname="q1", + print_contents=[PrintContent(object_key="doc-id", document_name="f.pdf")], + ) + client = PrintClient(mock_http) + client.create_print_task(task) + + _, kwargs = mock_http.put.call_args + assert kwargs["json"]["username"] == "auto@example.com" + assert task.username == "auto@example.com" + + def test_username_not_overwritten_when_provided(self): + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + mock_http.get_username.return_value = "auto@example.com" + + task = PrintTask( + item_id="doc-id", + qname="q1", + print_contents=[PrintContent(object_key="doc-id", document_name="f.pdf")], + username="explicit@example.com", + ) + client = PrintClient(mock_http) + client.create_print_task(task) + + _, kwargs = mock_http.put.call_args + assert kwargs["json"]["username"] == "explicit@example.com" + mock_http.get_username.assert_not_called() + + def test_http_error_raises_operation_error(self): + mock_http = MagicMock() + mock_http.put.side_effect = HttpError("rate limited", status_code=429) + + task = PrintTask( + item_id="doc-id", + qname="q1", + print_contents=[PrintContent(object_key="doc-id", document_name="f.pdf")], + ) + client = PrintClient(mock_http) + with pytest.raises(PrintOperationError, match="failed to create print task"): + client.create_print_task(task) + + +class TestPrintTaskMetadata: + + def test_to_dict_includes_all_fields(self): + from sap_cloud_sdk.print._models import PrintTaskMetadata + meta = PrintTaskMetadata(version=1.0, business_user="user@example.com", object_node_type="Invoice") + result = meta.to_dict() + assert result["version"] == 1.0 + assert result["business_metadata"]["business_user"] == "user@example.com" + assert result["business_metadata"]["object_node_type"] == "Invoice" + + def test_create_print_task_body_includes_metadata(self): + from sap_cloud_sdk.print._models import PrintTaskMetadata + mock_http = MagicMock() + mock_http.put.return_value = _mock_response(204) + + meta = PrintTaskMetadata(version=1.0, business_user="user@example.com") + task = PrintTask( + item_id="doc-id", + qname="q1", + print_contents=[PrintContent(object_key="doc-id", document_name="f.pdf")], + username="user@example.com", + metadata=meta, + ) + client = PrintClient(mock_http) + client.create_print_task(task) + + _, kwargs = mock_http.put.call_args + assert "metadata" in kwargs["json"] + assert kwargs["json"]["metadata"]["version"] == 1.0 diff --git a/tests/print/unit/test_config.py b/tests/print/unit/test_config.py new file mode 100644 index 0000000..da63645 --- /dev/null +++ b/tests/print/unit/test_config.py @@ -0,0 +1,116 @@ +"""Unit tests for print module config loading.""" + +import json +import pytest +from unittest.mock import patch + +from sap_cloud_sdk.print.config import PrintConfig, load_from_env_or_mount +from sap_cloud_sdk.print.exceptions import ConfigError + + +def _uaa_json( + clientid="cid", clientsecret="csecret", url="https://auth.example.com" +) -> str: + return json.dumps({"clientid": clientid, "clientsecret": clientsecret, "url": url}) + + +class TestLoadFromEnvOrMount: + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_returns_print_config(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "https://api.eu10.print.services.sap" + target.uaa = _uaa_json() + + mock_resolver.side_effect = fill_binding + + config = load_from_env_or_mount() + + assert isinstance(config, PrintConfig) + assert config.url == "https://api.eu10.print.services.sap" + assert config.client_id == "cid" + assert config.client_secret == "csecret" + assert config.token_url == "https://auth.example.com/oauth/token" + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_uses_default_instance(self, mock_resolver): + calls = [] + + def capture(**kwargs): + calls.append(kwargs) + kwargs["target"].url = "https://api.eu10.print.services.sap" + kwargs["target"].uaa = _uaa_json() + + mock_resolver.side_effect = capture + + load_from_env_or_mount() + assert calls[0]["instance"] == "default" + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_uses_provided_instance(self, mock_resolver): + calls = [] + + def capture(**kwargs): + calls.append(kwargs) + kwargs["target"].url = "https://api.eu10.print.services.sap" + kwargs["target"].uaa = _uaa_json() + + mock_resolver.side_effect = capture + + load_from_env_or_mount(instance="prod") + assert calls[0]["instance"] == "prod" + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_missing_url_raises_config_error(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "" + target.uaa = _uaa_json() + + mock_resolver.side_effect = fill_binding + + with pytest.raises(ConfigError, match="failed to load print configuration"): + load_from_env_or_mount() + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_invalid_uaa_json_raises_config_error(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "https://api.eu10.print.services.sap" + target.uaa = "not-valid-json" + + mock_resolver.side_effect = fill_binding + + with pytest.raises(ConfigError): + load_from_env_or_mount() + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_missing_clientid_raises_config_error(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "https://api.eu10.print.services.sap" + target.uaa = _uaa_json(clientid="") + + mock_resolver.side_effect = fill_binding + + with pytest.raises(ConfigError, match="clientid"): + load_from_env_or_mount() + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_missing_clientsecret_raises_config_error(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "https://api.eu10.print.services.sap" + target.uaa = _uaa_json(clientsecret="") + + mock_resolver.side_effect = fill_binding + + with pytest.raises(ConfigError, match="clientsecret"): + load_from_env_or_mount() + + @patch("sap_cloud_sdk.print.config.read_from_mount_and_fallback_to_env_var") + def test_missing_uaa_url_raises_config_error(self, mock_resolver): + def fill_binding(*, target, **_): + target.url = "https://api.eu10.print.services.sap" + target.uaa = _uaa_json(url="") + + mock_resolver.side_effect = fill_binding + + with pytest.raises(ConfigError, match="uaa.url"): + load_from_env_or_mount() diff --git a/tests/print/unit/test_http.py b/tests/print/unit/test_http.py new file mode 100644 index 0000000..c10c706 --- /dev/null +++ b/tests/print/unit/test_http.py @@ -0,0 +1,223 @@ +"""Unit tests for Print HTTP transport and TokenProvider.""" + +import base64 +import json +import pytest +from unittest.mock import MagicMock, patch +from requests.exceptions import RequestException + +from sap_cloud_sdk.print._http import TokenProvider, PrintHttp +from sap_cloud_sdk.print.config import PrintConfig +from sap_cloud_sdk.print.exceptions import HttpError + + +def _make_jwt(claims: dict) -> str: + """Build a minimal unsigned JWT with the given claims.""" + header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=").decode() + return f"{header}.{payload}." + + +def _config() -> PrintConfig: + return PrintConfig( + url="https://api.eu10.print.services.sap", + token_url="https://tenant.authentication.eu10.hana.ondemand.com/oauth/token", + client_id="client-id", + client_secret="client-secret", + ) + + +class TestTokenProvider: + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_returns_access_token(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = {"access_token": "tok-abc"} + + provider = TokenProvider(_config()) + assert provider.get_token() == "tok-abc" + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_missing_access_token_raises(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = {"expires_in": 3600} + + provider = TokenProvider(_config()) + with pytest.raises(HttpError, match="missing access_token"): + provider.get_token() + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_resolve_username_returns_user_name_claim(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = { + "access_token": _make_jwt({"user_name": "john.doe@example.com", "client_id": "sb-app"}) + } + + provider = TokenProvider(_config()) + assert provider.resolve_username() == "john.doe@example.com" + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_resolve_username_falls_back_to_client_id_claim(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = { + "access_token": _make_jwt({"client_id": "sb-app!t123"}) + } + + provider = TokenProvider(_config()) + assert provider.resolve_username() == "sb-app!t123" + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_resolve_username_falls_back_to_config_client_id_on_bad_token(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = {"access_token": "not.a.jwt"} + + provider = TokenProvider(_config()) + assert provider.resolve_username() == "client-id" + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_resolve_username_uses_cached_token(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.return_value = { + "access_token": _make_jwt({"user_name": "cached@example.com"}) + } + + provider = TokenProvider(_config()) + provider.get_token() + # fetch_token should not be called again + mock_session.fetch_token.reset_mock() + assert provider.resolve_username() == "cached@example.com" + mock_session.fetch_token.assert_not_called() + + +class TestPrintHttp: + + def _http(self, mock_session) -> PrintHttp: + config = _config() + mock_tp = MagicMock() + mock_tp.get_token.return_value = "test-token" + return PrintHttp(config=config, token_provider=mock_tp, session=mock_session) + + def _ok_response(self, status_code: int = 200): + resp = MagicMock() + resp.status_code = status_code + resp.text = "" + return resp + + def _error_response(self, status_code: int): + resp = MagicMock() + resp.status_code = status_code + resp.text = "error body" + return resp + + def test_get_constructs_correct_url(self): + mock_session = MagicMock() + mock_session.request.return_value = self._ok_response() + + http = self._http(mock_session) + http.get("qm/api/v1/rest/queues") + + _, kwargs = mock_session.request.call_args + assert "print.services.sap" in kwargs["url"] + assert "queues" in kwargs["url"] + assert kwargs["method"] == "GET" + + def test_authorization_header_set(self): + mock_session = MagicMock() + mock_session.request.return_value = self._ok_response() + + http = self._http(mock_session) + http.get("some/path") + + _, kwargs = mock_session.request.call_args + assert kwargs["headers"]["Authorization"] == "Bearer test-token" + + def test_non_2xx_raises_http_error(self): + mock_session = MagicMock() + mock_session.request.return_value = self._error_response(500) + + http = self._http(mock_session) + with pytest.raises(HttpError) as exc_info: + http.get("some/path") + assert exc_info.value.status_code == 500 + + def test_request_exception_raises_http_error(self): + mock_session = MagicMock() + mock_session.request.side_effect = RequestException("connection refused") + + http = self._http(mock_session) + with pytest.raises(HttpError, match="request failed"): + http.get("some/path") + + def test_put_sends_json_body(self): + mock_session = MagicMock() + mock_session.request.return_value = self._ok_response(204) + + http = self._http(mock_session) + http.put("some/path", json={"key": "value"}) + + _, kwargs = mock_session.request.call_args + assert kwargs["json"] == {"key": "value"} + assert kwargs["method"] == "PUT" + + def test_post_multipart_sends_files(self): + mock_session = MagicMock() + mock_session.request.return_value = self._ok_response(201) + + http = self._http(mock_session) + http.post("some/path", files={"file": ("doc.pdf", b"data")}) + + _, kwargs = mock_session.request.call_args + assert kwargs["files"] is not None + assert kwargs["method"] == "POST" + + def test_extra_headers_merged_into_request(self): + mock_session = MagicMock() + mock_session.request.return_value = self._ok_response() + + http = self._http(mock_session) + http.get("some/path", headers={"X-Custom": "value"}) + + _, kwargs = mock_session.request.call_args + assert kwargs["headers"]["X-Custom"] == "value" + assert kwargs["headers"]["Authorization"] == "Bearer test-token" + + def test_response_text_read_failure_still_raises_http_error(self): + mock_session = MagicMock() + resp = MagicMock() + resp.status_code = 500 + type(resp).text = property(lambda self: (_ for _ in ()).throw(RuntimeError("unreadable"))) + mock_session.request.return_value = resp + + http = self._http(mock_session) + with pytest.raises(HttpError) as exc_info: + http.get("some/path") + assert exc_info.value.status_code == 500 + + def test_get_username_delegates_to_token_provider(self): + mock_session = MagicMock() + config = _config() + mock_tp = MagicMock() + mock_tp.resolve_username.return_value = "user@example.com" + http = PrintHttp(config=config, token_provider=mock_tp, session=mock_session) + + assert http.get_username() == "user@example.com" + mock_tp.resolve_username.assert_called_once() + + +class TestTokenProviderFetchFailure: + + @patch("sap_cloud_sdk.print._http.OAuth2Session") + def test_fetch_token_exception_raises_http_error(self, mock_oauth): + mock_session = MagicMock() + mock_oauth.return_value = mock_session + mock_session.fetch_token.side_effect = Exception("(invalid_client) Bad credentials") + + provider = TokenProvider(_config()) + with pytest.raises(HttpError, match="failed to acquire token"): + provider.get_token() diff --git a/uv.lock b/uv.lock index 55b94cc..75bc37a 100644 --- a/uv.lock +++ b/uv.lock @@ -3682,7 +3682,7 @@ wheels = [ [[package]] name = "sap-cloud-sdk" -version = "0.23.2" +version = "0.24.1" source = { editable = "." } dependencies = [ { name = "grpcio" },