diff --git a/.env_integration_tests.example b/.env_integration_tests.example index 54e44a28..ebf9b605 100644 --- a/.env_integration_tests.example +++ b/.env_integration_tests.example @@ -39,6 +39,16 @@ AICORE_FILTER_TEST_SELF_HARM_PROMPT= CLOUD_SDK_CFG_AUDITLOG_DEFAULT_URL=https://your-auditlog-api-url-here CLOUD_SDK_CFG_AUDITLOG_DEFAULT_UAA='{"url":"https://your-auth-url","clientid":"your-client-id","clientsecret":"your-client-secret"}' +# CBC (Central Business Configuration) +# Resolves the CBC URL from the tenant-mapping Destination Fragment and the mTLS +# certificate from the Destination Service, so it also needs the DESTINATION +# binding vars below. The three CLOUD_SDK_CFG_CBC_DEFAULT_* values seed the SDK +# runtime context (app tenant id, subscriber subdomain) and derive the +# certificate name from the landscape. +CLOUD_SDK_CFG_CBC_DEFAULT_APP_TENANT_ID=your-app-tenant-id-here +CLOUD_SDK_CFG_CBC_DEFAULT_TENANT_SUBDOMAIN=your-subscriber-tenant-subdomain-here +CLOUD_SDK_CFG_CBC_DEFAULT_LANDSCAPE=your-landscape-here + # DATA ANONYMIZATION CLOUD_SDK_CFG_DATA_ANONYMIZATION_DEFAULT_URL=https://your-data-anonymization-api-url-here CLOUD_SDK_CFG_DATA_ANONYMIZATION_DEFAULT_CERT=your-base64-encoded-client-certificate-pem diff --git a/.gitignore b/.gitignore index ae33b087..e632cb42 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,9 @@ env.bak/ venv.bak/ piperBuild-env/ +# Local scratch +tmp/ + # IDEs .vscode/ .idea/ diff --git a/src/sap_cloud_sdk/cbc/__init__.py b/src/sap_cloud_sdk/cbc/__init__.py new file mode 100644 index 00000000..58ac0177 --- /dev/null +++ b/src/sap_cloud_sdk/cbc/__init__.py @@ -0,0 +1,85 @@ +"""SAP Cloud SDK for Python — CBC (Central Business Configuration) module. + +Provides a typed Python client for reading tenant-specific business configuration +from SAP Central Business Configuration (CBC). + +CBC is an SAP service that manages tenant-specific business configuration for +SAP cloud applications and AI agents. + +Quick start:: + + from sap_cloud_sdk import cbc + + # Infers everything from provisioning: the app tenant id and subdomain come + # from the SDK runtime context (wire bootstrap(app) once at startup), the CBC + # URL from the tenant-mapping Destination Fragment, and the mTLS certificate + # from the Destination Service. + cbc_client = cbc.create_client() + config = cbc_client.get_configuration() + + # Access entity data + payment = config.get_configuration_object("payment-config") + if payment: + for row in payment.get_entity_data("payment-mode").as_list(): + print(row) + +To override any individual piece (e.g. for local development or tests), pass a +:class:`~sap_cloud_sdk.cbc.config.CBCConfig` — see the CBC user guide. +""" + +from __future__ import annotations + +from sap_cloud_sdk.cbc.client import ( + CBCClient, + create_client, +) +from sap_cloud_sdk.cbc.config import ( + CBCConfig, +) +from sap_cloud_sdk.cbc.exceptions import ( + CBCError, + CBCClientError, + CBCConfigError, + CBCHttpError, + CBCNetworkError, + CBCServerError, + HttpContext, +) +from sap_cloud_sdk.cbc._models import ( + ApiError, + ConfigurationData, + ConfigurationEntity, + ConfigurationObject, + ConsumptionVersion, + ConsumptionVersions, + EntityData, +) + + +__all__ = [ + # factory + "create_client", + # client + "CBCClient", + # configuration + "CBCConfig", + # exceptions + "CBCError", + "CBCClientError", + "CBCConfigError", + "CBCHttpError", + "CBCNetworkError", + "CBCServerError", + "HttpContext", + # models — consumption versions + "ConsumptionVersion", + "ConsumptionVersions", + # models — entities + "EntityData", + "ConfigurationEntity", + "ConfigurationObject", + # models — configuration + "ConfigurationData", + # models — api error + "ApiError", +] diff --git a/src/sap_cloud_sdk/cbc/_models.py b/src/sap_cloud_sdk/cbc/_models.py new file mode 100644 index 00000000..672c8df5 --- /dev/null +++ b/src/sap_cloud_sdk/cbc/_models.py @@ -0,0 +1,292 @@ +"""Data models for the CBC (Central Business Configuration) module. + +All models use Pydantic v2 with ``frozen=True`` and camelCase alias support +so they map directly to the CBC REST API JSON. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from typing import Any, cast +from pydantic import BaseModel, ConfigDict, Field + + +class _FrozenModel(BaseModel): + """Base model: immutable, accepts both snake_case and camelCase field names.""" + + model_config = ConfigDict(frozen=True, populate_by_name=True) + + +# --------------------------------------------------------------------------- +# Consumption version models +# --------------------------------------------------------------------------- + + +class ConsumptionVersion(_FrozenModel): + """A snapshot of the business configuration for an app tenant at a point in time. + + Attributes: + version: Version identifier. + created_date: Creation timestamp. + modified_date: Last modification timestamp. + """ + + version: str + created_date: datetime | None = Field(default=None, alias="createdDate") + modified_date: datetime | None = Field(default=None, alias="modifiedDate") + + +class ConsumptionVersions(_FrozenModel): + """Collection of consumption versions returned by the CBC API. + + Attributes: + items: List of :class:`ConsumptionVersion` objects. + """ + + items: list[ConsumptionVersion] + + def latest(self) -> ConsumptionVersion | None: + """Return the latest version. + + Prefers most-recent ``modifiedDate``; falls back to ``createdDate``; falls + back to last item in the list. + + Returns: + Latest :class:`ConsumptionVersion`, or ``None`` if the list is empty. + """ + if not self.items: + return None + dated = [v for v in self.items if v.modified_date is not None] + if dated: + return max(dated, key=lambda v: cast(datetime, v.modified_date)) + created = [v for v in self.items if v.created_date is not None] + if created: + return max(created, key=lambda v: cast(datetime, v.created_date)) + return self.items[-1] + + +# --------------------------------------------------------------------------- +# Entity models +# --------------------------------------------------------------------------- + + +class ConfigurationObjectEntity(_FrozenModel): + """One entity listed under a config object by the ``configurationObjects`` API. + + Attributes: + entity_id: Authored entity key (e.g. ``"payment-mode"``), used directly + in the entity-data URL path. + """ + + entity_id: str = Field(alias="entityId") + + +class ConfigurationObjectEntry(_FrozenModel): + """One config object and its entities as returned by the API. + + Attributes: + configuration_object_id: Authored config object identifier (e.g. ``"payment-config"``). + entities: Entities belonging to this config object. + """ + + configuration_object_id: str = Field(alias="configurationObjectId") + entities: list[ConfigurationObjectEntity] + + +class ConfigurationObjectList(_FrozenModel): + """Config objects for a consumption version, already grouped by the API. + + Attributes: + items: List of :class:`ConfigurationObjectEntry` objects. + """ + + items: list[ConfigurationObjectEntry] + + +class EntityData: + """Configuration data for an entity. + + Wraps the raw API response data and enforces shape at access time. + """ + + def __init__(self, raw: list[dict[str, Any]] | dict[str, Any]) -> None: + self._raw = raw + + def is_list(self) -> bool: + """Return ``True`` if the data is a list (``as_list()`` is safe to call).""" + return isinstance(self._raw, list) + + def is_object(self) -> bool: + """Return ``True`` if the data is a dict (``as_object()`` is safe to call).""" + return isinstance(self._raw, dict) + + def value(self) -> list[dict[str, Any]] | dict[str, Any]: + """Return the data as-is, without asserting its shape. + + Use :meth:`as_list` / :meth:`as_object` when you expect a specific shape, + or :meth:`is_list` / :meth:`is_object` to check first. + """ + return self._raw + + def as_list(self) -> list[dict[str, Any]]: + """Return the data as a list of objects. + + Raises: + ValueError: If the data is a dict, not a list. + """ + if not isinstance(self._raw, list): + raise ValueError( + "Entity data is a dict, not a list — use as_object() instead." + ) + return self._raw + + def as_object(self) -> dict[str, Any]: + """Return the data as a dict. + + Raises: + ValueError: If the data is a list, not a dict. + """ + if not isinstance(self._raw, dict): + raise ValueError( + "Entity data is a list, not a dict — use as_list() instead." + ) + return self._raw + + def __repr__(self) -> str: + return f"EntityData({self._raw!r})" + + +@dataclass +class ConfigurationEntity: + """A single entity within a config object. + + Attributes: + entity_id: Authored entity identifier (e.g. ``"payment-mode"``). + data: Configuration data for this entity. + """ + + entity_id: str + data: EntityData + + +@dataclass +class ConfigurationObject: + """A configuration object and its entities. + + Attributes: + configuration_object_id: Authored config object identifier (e.g. ``"payment-config"``). + entities: Entity data for all entities in this config object. + """ + + configuration_object_id: str + entities: list[ConfigurationEntity] + + def get_entity_data(self, entity_id: str) -> EntityData | None: + """Return the data for the entity with the given ID. + + Args: + entity_id: Authored entity identifier. + + Returns: + :class:`EntityData` for the matching entity, or ``None`` if not found. + """ + entity = next((e for e in self.entities if e.entity_id == entity_id), None) + return entity.data if entity is not None else None + + +@dataclass +class ConfigurationData: + """Complete business configuration — all config objects for one consumption version. + + Attributes: + consumption_version: Version this data was fetched from. + app_tenant_id: Application tenant this data belongs to. + configuration_objects: Configuration objects and their entity data. + """ + + consumption_version: str + app_tenant_id: str + configuration_objects: list[ConfigurationObject] + + def get_configuration_object( + self, configuration_object_id: str + ) -> ConfigurationObject | None: + """Return the config object with the given ID. + + Args: + configuration_object_id: Authored config object identifier. + + Returns: + Matching :class:`ConfigurationObject`, or ``None`` if not found. + """ + return next( + ( + co + for co in self.configuration_objects + if co.configuration_object_id == configuration_object_id + ), + None, + ) + + def get_entity_data( + self, configuration_object_id: str, entity_id: str + ) -> EntityData | None: + """Return the data for the given config object and entity IDs. + + Args: + configuration_object_id: Authored config object identifier. + entity_id: Authored entity identifier. + + Returns: + :class:`EntityData` for the matching entity, or ``None`` if not found. + """ + co = self.get_configuration_object(configuration_object_id) + return co.get_entity_data(entity_id) if co is not None else None + + +# --------------------------------------------------------------------------- +# API error model +# --------------------------------------------------------------------------- + + +class ApiError(_FrozenModel): + """Error payload returned by the CBC API. + + Attributes: + code: Application-level error code. + message: Human-readable error message. + """ + + code: str + message: str + + @classmethod + def from_response(cls, response_body: bytes | None) -> "ApiError": + """Parse a CBC API error response body. + + Falls back to a generic ``UNKNOWN_ERROR`` when the body is absent or + unparseable. + + Args: + response_body: Raw HTTP response body. + + Returns: + Parsed :class:`ApiError`. + """ + if not response_body: + return cls(code="UNKNOWN_ERROR", message="No response body") + try: + data = json.loads(response_body.decode("utf-8")) + if isinstance(data, dict) and "error" in data: + return cls( + code=data["error"].get("code", "UNKNOWN_ERROR"), + message=data["error"].get("message", "Unknown error"), + ) + except (json.JSONDecodeError, UnicodeDecodeError): + pass + return cls( + code="UNKNOWN_ERROR", + message=response_body.decode("utf-8", errors="replace").strip(), + ) diff --git a/src/sap_cloud_sdk/cbc/client.py b/src/sap_cloud_sdk/cbc/client.py new file mode 100644 index 00000000..fa7ceea5 --- /dev/null +++ b/src/sap_cloud_sdk/cbc/client.py @@ -0,0 +1,759 @@ +"""CBC client for reading business configuration from CBC. + +This module provides: + +- :class:`CBCClient` — the CBC client; reads tenant-specific business + configuration from the CBC service over mTLS. +- :func:`create_client` — the factory. Call it with no arguments to infer + everything from the application-platform provisioning conventions, or pass a + :class:`~sap_cloud_sdk.cbc.config.CBCConfig` to override individual pieces. + +By default the client derives, per request, the CBC URL and the application +tenant id from the SDK-wide runtime context +(:mod:`sap_cloud_sdk.core.runtime_context`, populated once at startup by +``bootstrap(app)``), and loads the app's provider-level mTLS certificate from +the Destination Service. The certificate is reloaded automatically on a TLS +handshake failure, so a long-lived client recovers from certificate rotation +without being recreated. + +Quick start:: + + from sap_cloud_sdk import cbc + + cbc_client = cbc.create_client() # infer everything from provisioning + config = cbc_client.get_configuration() +""" + +from __future__ import annotations + +import logging +import os +import ssl +import threading +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +import httpx + +from sap_cloud_sdk.cbc._models import ( + ApiError, + ConfigurationData, + ConfigurationEntity, + ConfigurationObject, + ConfigurationObjectList, + ConsumptionVersions, + EntityData, +) +from sap_cloud_sdk.cbc.config import CBCConfig +from sap_cloud_sdk.cbc.exceptions import ( + CBCClientError, + CBCConfigError, + CBCNetworkError, + CBCServerError, + HttpContext, +) +from sap_cloud_sdk.core.runtime_context import ( + APP_TENANT_ID, + DWC_SUBDOMAIN, + get_context, +) +from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Platform provisioning conventions +# --------------------------------------------------------------------------- +# The fragment prefix and certificate name below follow the application-platform +# convention; they are not arbitrary. Override via CBCConfig / env vars if your +# setup differs. + +#: Prefix of the Destination Fragment that maps a subdomain to its CBC tenant +#: (platform convention). +CBC_FRAGMENT_PREFIX = "CBC_TenantMapping_" + +#: Template for the app's own provider-level mTLS certificate name +#: (platform convention). +_CERT_NAME_TEMPLATE = "sap-managed-runtime-ias-{landscape}.pem" + +# Env overrides — the defaults cover the common case; set these only to override. + +#: The ``instance`` passed to the destination ``create_fragment_client`` / +#: ``create_certificate_client`` (used for secret resolution in cloud mode, +#: defaults to "default"); it is not the individual destination / cert / +#: fragment, which are selected by name. +ENV_DESTINATION_INSTANCE = "CLOUD_SDK_CBC_DESTINATION_INSTANCE" + +#: Explicit certificate name, overriding the landscape-derived default. +ENV_CERT_NAME = "CLOUD_SDK_CBC_CERTIFICATE_NAME" + +#: Password for the certificate keystore, if encrypted (default: none). +ENV_P12_PASSWORD = "CLOUD_SDK_CBC_P12_PASSWORD" + +#: Platform landscape, used to derive the default certificate name. +ENV_LANDSCAPE = "APPFND_CONHOS_LANDSCAPE" + + +def _is_tls_failure(exc: BaseException) -> bool: + """Return ``True`` if an :class:`ssl.SSLError` appears in the cause chain. + + An expired or rotated mTLS client certificate is rejected by the server at + the TLS handshake, which httpx surfaces as a transport error (e.g. + ``httpx.ReadError``) wrapping ``httpcore.ReadError`` wrapping + ``ssl.SSLError``. The ``ssl.SSLError`` is not the top-level type, so walk + the ``__cause__`` / ``__context__`` chain instead of checking ``isinstance`` + on ``exc`` directly. ``seen`` guards against a cyclic chain. + """ + seen: set[int] = set() + cur: BaseException | None = exc + while cur is not None and id(cur) not in seen: + seen.add(id(cur)) + if isinstance(cur, ssl.SSLError): + return True + cur = cur.__cause__ or cur.__context__ + return False + + +# --------------------------------------------------------------------------- +# Internal config dataclass +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class _ApiPaths: + """API path configuration for a :class:`CBCClient` instance.""" + + configurations_path: str + + +# --------------------------------------------------------------------------- +# CBCClient +# --------------------------------------------------------------------------- + + +class CBCClient: + """Client for reading business configuration from CBC. + + Prefer :func:`create_client` over direct instantiation:: + + cbc_client = cbc.create_client() + + Direct instantiation is supported for testing (inject a mock ``http_client``). + + **Blocking I/O.** All public methods make synchronous HTTP calls via + ``httpx.Client``. In an async context, wrap calls with + ``asyncio.to_thread(client.get_configuration)`` to avoid blocking the event + loop. + + **Thread safety.** Safe to share a single instance across threads and async + tasks. The internal ``httpx.Client`` is thread-safe. The only mutable state + is the cached ``ssl.SSLContext`` — the app's own provider-level mTLS + certificate, which is deployment/landscape-specific and shared across all + tenants. It is rebuilt under a lock on TLS handshake failure — concurrent + rebuilds are safe and at most one rebuild runs at a time. + + Args: + base_url: Callable returning the CBC service base URL. Invoked on every + request — in a multi-tenant agent the URL comes from a request-scoped + Destination Fragment, so it is resolved per call. + app_tenant_id: Callable returning the application tenant identifier. + Invoked on every request and sent as the ``appTenantId`` query + parameter. + http_client: Optional pre-configured ``httpx.Client`` — takes full + precedence over ``ssl_context``. Use for testing. + ssl_context: Optional callable returning a freshly-built + :class:`ssl.SSLContext` with mTLS loaded. Resolved **once** at + construction, then re-invoked **only** on a TLS handshake failure to + pick up a rotated certificate (not per request). ``None`` for the + plain-HTTP / local-mock path. + """ + + def __init__( + self, + base_url: Callable[[], str], + app_tenant_id: Callable[[], str], + http_client: httpx.Client | None = None, + ssl_context: Callable[[], ssl.SSLContext] | None = None, + ) -> None: + self._base_url = base_url + self._app_tenant_id = app_tenant_id + self._config = _ApiPaths( + configurations_path="/configuration/v1", + ) + self._ssl_factory = ssl_context + self._lock = threading.Lock() + self._http_client = http_client or self._build_http_client() + + def _build_http_client(self) -> httpx.Client: + """Build an ``httpx.Client``, resolving the SSL context from the factory. + + ``verify`` is either the resolved mTLS :class:`ssl.SSLContext` or ``True`` + (httpx's default CA bundle) — never ``False``, so TLS verification is + always on. + """ + ctx = self._ssl_factory() if self._ssl_factory else None + return httpx.Client(verify=ctx if ctx is not None else True) + + def close(self) -> None: + """Close the underlying HTTP client and release connections.""" + self._http_client.close() + + def __enter__(self) -> "CBCClient": + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + def _resolve_app_tenant_id(self) -> str: + return self._app_tenant_id() + + # ------------------------------------------------------------------ + # Public API methods + # ------------------------------------------------------------------ + + @record_metrics(Module.CBC, Operation.CBC_GET_CONSUMPTION_VERSIONS) + def get_consumption_versions(self) -> ConsumptionVersions: + """Return available consumption versions for the configured tenant. + + A consumption version represents a snapshot of the business configuration + for an app tenant at a point in time. Use this to discover the active + version ID when you don't already have it. + + Returns: + :class:`ConsumptionVersions` with all versions for the tenant. + + Raises: + CBCClientError: On 4xx responses. + CBCServerError: On 5xx responses. + CBCNetworkError: On connection failures. + """ + return self._get_consumption_versions( + self._base_url(), self._resolve_app_tenant_id() + ) + + def _get_consumption_versions( + self, base_url: str, app_tenant_id: str + ) -> ConsumptionVersions: + """Return the consumption versions, using already-resolved request values. + + Takes ``base_url`` / ``app_tenant_id`` as arguments so a caller that has + already resolved them (e.g. :meth:`get_configuration`) does not resolve + them a second time — resolution can hit the Destination Service, and a + second resolve could also disagree with the first if the request context + changed in between. + + Args: + base_url: Resolved CBC service base URL for this operation. + app_tenant_id: Application tenant identifier. + + Raises: + CBCClientError: On 4xx responses. + CBCServerError: On 5xx responses. + CBCNetworkError: On connection failures. + """ + url = self._configurations_url( + base_url, + f"/consumptionVersions?appTenantId={app_tenant_id}", + ) + return ConsumptionVersions.model_validate(self._request("GET", url).json()) + + def _get_configuration_objects( + self, base_url: str, app_tenant_id: str, consumption_version: str + ) -> ConfigurationObjectList: + """Return the config objects (with their entities) for the given version. + + The API returns config objects already grouped with their child entities, + so no client-side grouping is needed. + + Args: + base_url: Resolved CBC service base URL for this operation. + app_tenant_id: Application tenant identifier. + consumption_version: Consumption version ID. + + Returns: + :class:`ConfigurationObjectList` containing config objects and their entities. + + Raises: + CBCClientError: On 4xx responses. + CBCServerError: On 5xx responses. + CBCNetworkError: On connection failures. + """ + url = self._configurations_url( + base_url, + f"/consumptionVersions/{consumption_version}/configurationObjects" + f"?appTenantId={app_tenant_id}", + ) + return ConfigurationObjectList.model_validate(self._request("GET", url).json()) + + @record_metrics(Module.CBC, Operation.CBC_GET_CONFIGURATION) + def get_configuration( + self, + consumption_version: str | None = None, + ) -> ConfigurationData: + """Return the full business configuration for the configured tenant. + + Fetches the config objects and the data for every entity they contain, for + the specified consumption version. When ``consumption_version`` is omitted, + the latest version is resolved automatically via + :meth:`get_consumption_versions`. + + Args: + consumption_version: Consumption version ID. When ``None``, the + latest version is resolved via :meth:`get_consumption_versions`. + + Returns: + :class:`ConfigurationData` containing all entity data for the version. + + Raises: + CBCClientError: On 4xx responses, or when no consumption version exists + for the tenant and ``consumption_version`` was not provided. + CBCServerError: On 5xx responses. + CBCNetworkError: On connection failures. + """ + base_url = self._base_url() + app_tenant_id = self._resolve_app_tenant_id() + if consumption_version is None: + versions = self._get_consumption_versions(base_url, app_tenant_id) + latest = versions.latest() + if latest is None: + raise CBCClientError( + f"CBC returned no consumption version for tenant={app_tenant_id!r}." + ) + consumption_version = latest.version + + co_list = self._get_configuration_objects( + base_url, app_tenant_id, consumption_version + ) + configuration_objects = [ + ConfigurationObject( + configuration_object_id=entry.configuration_object_id, + entities=[ + self._fetch_entity_data( + base_url, + app_tenant_id, + consumption_version, + entry.configuration_object_id, + entity.entity_id, + ) + for entity in entry.entities + ], + ) + for entry in co_list.items + ] + return ConfigurationData( + consumption_version=consumption_version, + app_tenant_id=app_tenant_id, + configuration_objects=configuration_objects, + ) + + @record_metrics(Module.CBC, Operation.CBC_GET_ENTITY_DATA) + def get_entity_data( + self, + configuration_object_id: str, + entity_id: str, + consumption_version: str | None = None, + ) -> EntityData: + """Return the data for a single entity without fetching all configuration. + + Makes a targeted HTTP call for the one entity, avoiding the per-entity + calls that :meth:`get_configuration` makes for every entity in every + config object. + + Args: + configuration_object_id: Authored config object identifier (e.g. ``"payment-config"``). + entity_id: Authored entity identifier (e.g. ``"payment-mode"``). + consumption_version: Consumption version ID. When ``None``, the + latest version is resolved via :meth:`get_consumption_versions`. + + Returns: + :class:`EntityData` for the entity. + + Raises: + CBCClientError: On 4xx responses, or when no consumption version exists + for the tenant and ``consumption_version`` was not provided. + CBCServerError: On 5xx responses. + CBCNetworkError: On connection failures. + """ + base_url = self._base_url() + app_tenant_id = self._resolve_app_tenant_id() + if consumption_version is None: + versions = self._get_consumption_versions(base_url, app_tenant_id) + latest = versions.latest() + if latest is None: + raise CBCClientError( + f"CBC returned no consumption version for tenant={app_tenant_id!r}." + ) + consumption_version = latest.version + return self._fetch_entity_data( + base_url, + app_tenant_id, + consumption_version, + configuration_object_id, + entity_id, + ).data + + def _fetch_entity_data( + self, + base_url: str, + app_tenant_id: str, + consumption_version: str, + configuration_object_id: str, + entity_id: str, + ) -> ConfigurationEntity: + url = self._configurations_url( + base_url, + f"/consumptionVersions/{consumption_version}/configurationObjects" + f"/{configuration_object_id}/entities/{entity_id}/data" + f"?appTenantId={app_tenant_id}", + ) + body = self._request("GET", url).json() + + content = body.get("content", {}) if isinstance(body, dict) else {} + shape = body.get("contentShape") if isinstance(body, dict) else None + if shape == "OBJECT": + raw_data: list[dict[str, Any]] | dict[str, Any] = content.get("item", {}) + else: + # ARRAY / UNSPECIFIED / absent — items may be absent or empty. + raw_data = content.get("items", []) + + return ConfigurationEntity(entity_id=entity_id, data=EntityData(raw_data)) + + def _configurations_url(self, base_url: str, path: str = "") -> str: + base = base_url.rstrip("/") + return f"{base}{self._config.configurations_path}{path}" + + def _request( + self, + method: str, + url: str, + body: dict[str, Any] | None = None, + ) -> httpx.Response: + logger.debug("CBC %s %s", method, url) + try: + response = self._send(method, url, body) + except httpx.TransportError as exc: + # A TLS handshake failure (e.g. an expired/rotated client cert) is + # recoverable: rebuild the client from a fresh SSL context once and + # retry this one request. Non-TLS transport errors and all other + # request errors fall through to the terminal handler below. + if self._ssl_factory is not None and _is_tls_failure(exc): + logger.info("CBC TLS failure; reloading certificate and retrying") + self._rebuild_http_client() + try: + response = self._send(method, url, body) + except httpx.RequestError as retry_exc: + raise self._network_error(method, url, retry_exc) from retry_exc + else: + raise self._network_error(method, url, exc) from exc + except httpx.RequestError as exc: + raise self._network_error(method, url, exc) from exc + + if response.status_code >= 400: + ctx = HttpContext( + status_code=response.status_code, + request_method=method, + request_url=url, + ) + error = ApiError.from_response(response.content) + exc_class = ( + CBCServerError if response.status_code >= 500 else CBCClientError + ) + raise exc_class(error.message, code=error.code, http_context=ctx) + + return response + + def _send( + self, method: str, url: str, body: dict[str, Any] | None + ) -> httpx.Response: + """Perform one HTTP request via the underlying client. + + A thin wrapper over ``self._http_client.request`` with no error translation — + callers map failures to CBC exceptions and own any retry logic. + """ + return self._http_client.request( + method=method, + url=url, + headers={}, + json=body, + timeout=30.0, + ) + + def _rebuild_http_client(self) -> None: + """Rebuild the HTTP client from a fresh SSL context, under a lock. + + Called on a TLS failure to pick up a rotated certificate. Guards against + a thundering herd — if another thread already rebuilt while this one + waited on the lock, reuse that client instead of rebuilding again. + + The new client is built into a local before it replaces ``self._http_client``, + so a failing factory (which raises :class:`CBCConfigError` — the cert + loader wraps every load-time failure) leaves ``self._http_client`` on the prior + working context and the error propagates to the caller. + """ + with self._lock: + previous = self._http_client + new_client = self._build_http_client() + self._http_client = new_client + previous.close() + + def _network_error(self, method: str, url: str, exc: Exception) -> CBCNetworkError: + return CBCNetworkError( + f"Network error calling CBC: {exc}", + http_context=HttpContext( + status_code=-1, request_method=method, request_url=url + ), + ) + + +# --------------------------------------------------------------------------- +# Default platform resolvers +# --------------------------------------------------------------------------- +# These encode the SAP application-platform provisioning convention: the app +# tenant id and subdomain come from the SDK-wide runtime context (populated at +# startup by bootstrap(app)); the CBC URL comes from the tenant-mapping Fragment +# written during provisioning; the mTLS certificate comes from the Destination +# Service. create_client wires these in whenever CBCConfig leaves the matching +# field unset. + + +def _resolve_app_tenant_id() -> str: + """Return the application tenant id from the SDK runtime context. + + Reads :data:`~sap_cloud_sdk.core.runtime_context.APP_TENANT_ID` (the IAS + ``app_tid`` claim, populated by ``bootstrap(app)``). The application tenant + id identifies the subscriber tenant to CBC and is sent as the ``appTenantId`` + query parameter. + + Raises: + CBCConfigError: If no app tenant id is present in the runtime context. + """ + app_tid = get_context().get(APP_TENANT_ID) + if not app_tid: + raise CBCConfigError( + "No app tenant id in the runtime context — ensure bootstrap(app) is " + "wired so the IAS app_tid claim is populated, or override app_tenant_id " + "on CBCConfig." + ) + return app_tid + + +def _resolve_base_url(destination_instance: str) -> str: + """Return the CBC base URL for the current tenant. + + Reads the subscriber subdomain from the SDK runtime context + (:data:`~sap_cloud_sdk.core.runtime_context.DWC_SUBDOMAIN`), fetches the + tenant-mapping Fragment for that subdomain directly by name + (``CBC_TenantMapping_``), and returns its ``cbcUrl`` property + verbatim. + + Args: + destination_instance: The ``instance`` passed to the destination + ``create_fragment_client`` (selects which Destination Service binding + to read, used for secret resolution in cloud mode). + + Raises: + CBCConfigError: If the subdomain is absent from the runtime context, no + matching fragment is found, or the matched fragment has no ``cbcUrl`` + property. + """ + from sap_cloud_sdk.destination import AccessStrategy, create_fragment_client + from sap_cloud_sdk.destination.exceptions import DestinationError + + subdomain = get_context().get(DWC_SUBDOMAIN) + if not subdomain: + raise CBCConfigError( + "No subdomain in the runtime context — ensure bootstrap(app) is wired " + "so the dwc-subdomain is populated, or override base_url on CBCConfig." + ) + + fragment_name = f"{CBC_FRAGMENT_PREFIX}{subdomain}" + try: + client = create_fragment_client(instance=destination_instance) + fragment = client.get_subaccount_fragment( + fragment_name, + access_strategy=AccessStrategy.SUBSCRIBER_ONLY, + tenant=subdomain, + ) + except DestinationError as exc: + raise CBCConfigError( + f"Could not resolve the CBC URL: fetching tenant-mapping fragment " + f"{fragment_name!r} failed for subdomain={subdomain!r}: {exc}" + ) from exc + if fragment is None: + raise CBCConfigError( + f"No CBC mapping fragment for subdomain={subdomain!r}: " + f"{fragment_name!r} not found." + ) + cbc_url = fragment.properties.get("cbcUrl") + if not cbc_url: + raise CBCConfigError( + f"CBC mapping fragment {fragment.name!r} has no 'cbcUrl' property." + ) + return cbc_url + + +def _load_ssl_context( + destination_instance: str, cert_name: str, p12_password: bytes | None +) -> ssl.SSLContext: + """Fetch the provider certificate and load it into an :class:`ssl.SSLContext`. + + Args: + destination_instance: The ``instance`` passed to the destination + ``create_certificate_client`` (selects which Destination Service + binding to read, used for secret resolution in cloud mode). + cert_name: Name of the subaccount certificate to fetch. + p12_password: Certificate keystore password, or ``None`` if unencrypted. + + Raises: + CBCConfigError: If the certificate cannot be fetched or is not found. + """ + from sap_cloud_sdk.destination import AccessStrategy, create_certificate_client + from sap_cloud_sdk.destination._cert_loader import _load_pem + from sap_cloud_sdk.destination.exceptions import DestinationError + + try: + cert = create_certificate_client( + instance=destination_instance + ).get_subaccount_certificate( + cert_name, access_strategy=AccessStrategy.PROVIDER_ONLY + ) + except DestinationError as exc: + raise CBCConfigError( + f"Could not fetch the mTLS certificate {cert_name!r} from Destination " + f"Service instance {destination_instance!r}: {exc}" + ) from exc + if cert is None: + raise CBCConfigError( + f"Subaccount certificate {cert_name!r} not found in Destination " + f"Service instance {destination_instance!r}." + ) + return _load_pem(cert.content, p12_password, cert.name) + + +def _default_cert_name() -> str: + """Return the name of the app's provider-level mTLS certificate. + + This is the certificate the app presents to CBC for the mTLS handshake; its + name follows the platform convention :data:`_CERT_NAME_TEMPLATE`, with the + landscape filled in from :data:`ENV_LANDSCAPE`. + + Raises: + CBCConfigError: If :data:`ENV_LANDSCAPE` is not set, so the name cannot + be derived. + """ + landscape = os.environ.get(ENV_LANDSCAPE) + if not landscape: + raise CBCConfigError( + f"Cannot derive the CBC certificate name: {ENV_LANDSCAPE} is not set. " + f"Set it, or pass cbc_cert_name on CBCConfig." + ) + return _CERT_NAME_TEMPLATE.format(landscape=landscape) + + +def _cert_password_from_env() -> bytes | None: + """Return the cert keystore password from :data:`ENV_P12_PASSWORD`, or ``None``.""" + raw = os.environ.get(ENV_P12_PASSWORD) + return raw.encode() if raw else None + + +def _as_callable(value: str | Callable[[], str]) -> Callable[[], str]: + """Return ``value`` as a callable — wrap a plain string in a lambda.""" + if isinstance(value, str): + return lambda: value + return value + + +# --------------------------------------------------------------------------- +# Factory function +# --------------------------------------------------------------------------- + + +def create_client(config: CBCConfig | None = None) -> CBCClient: + """Create a CBC client. + + Called with no arguments, the client infers everything from the SAP + application-platform provisioning conventions: + + - ``app_tenant_id`` and the subscriber subdomain come from the SDK-wide + runtime context (populated once at startup by ``bootstrap(app)``); + - the CBC URL comes from the tenant-mapping Destination Fragment written + during provisioning; + - the mTLS certificate comes from the Destination Service, reloaded + automatically on a TLS handshake failure (certificate rotation). + + Pass a :class:`~sap_cloud_sdk.cbc.config.CBCConfig` to override any individual + piece. Each override wins over its ``CLOUD_SDK_CBC_*`` env var, which in turn + falls back to the platform default. Note that the mTLS certificate loader is + the default **only** in pure provisioning mode: once ``base_url`` or + ``app_tenant_id`` is overridden (e.g. pointing at a local mock), an unset + ``ssl_context`` means plain HTTPS — set ``ssl_context`` explicitly if you still + need a client certificate. + + Args: + config: Optional overrides. Leave ``None`` to infer everything from + provisioning. Set ``base_url`` / ``app_tenant_id`` / ``ssl_context`` + to bypass the corresponding default resolver; set + ``destination_instance`` / ``cbc_cert_name`` / ``p12_password`` to + steer the default certificate / fragment lookup. See + :class:`~sap_cloud_sdk.cbc.config.CBCConfig`. + + Returns: + A configured :class:`CBCClient`. + + Raises: + CBCConfigError: If the certificate name cannot be resolved, or — when a + request is made — if the certificate cannot be fetched, the runtime + context is missing the tenant id / subdomain, or the tenant-mapping + fragment is missing. + """ + if config is None: + config = CBCConfig() + + destination_instance = config.destination_instance + if destination_instance is None: + destination_instance = os.environ.get(ENV_DESTINATION_INSTANCE, "default") + + # app_tenant_id — explicit override else the runtime-context resolver. + if config.app_tenant_id is not None: + app_tenant_id = _as_callable(config.app_tenant_id) + else: + app_tenant_id = _resolve_app_tenant_id + + # base_url — explicit override else the tenant-mapping fragment resolver. + if config.base_url is not None: + base_url = _as_callable(config.base_url) + else: + base_url = lambda: _resolve_base_url(destination_instance) # noqa: E731 + + # ssl_context — explicit override else the Destination-Service cert loader. + # The mTLS loader is the default only in pure provisioning mode (neither + # base_url nor app_tenant_id overridden). Once a caller supplies either axis + # — e.g. pointing at a local mock — an unset ssl_context means plain HTTPS + # (public CA bundle, no client cert), so they are not forced to also set the + # platform landscape env or carry a certificate. The default loader resolves + # the certificate name/password lazily, inside the factory, consistent with + # the other default resolvers that only touch provisioning when invoked. + if config.ssl_context is not None: + ssl_context: Callable[[], ssl.SSLContext] | None = config.ssl_context + elif config.base_url is not None or config.app_tenant_id is not None: + ssl_context = None + else: + cfg_cert_name = config.cbc_cert_name + cfg_password = config.p12_password + + def ssl_context() -> ssl.SSLContext: + cert_name = ( + cfg_cert_name or os.environ.get(ENV_CERT_NAME) or _default_cert_name() + ) + password = ( + cfg_password if cfg_password is not None else _cert_password_from_env() + ) + return _load_ssl_context(destination_instance, cert_name, password) + + return CBCClient( + base_url=base_url, + app_tenant_id=app_tenant_id, + ssl_context=ssl_context, + ) diff --git a/src/sap_cloud_sdk/cbc/config.py b/src/sap_cloud_sdk/cbc/config.py new file mode 100644 index 00000000..cbdf39d6 --- /dev/null +++ b/src/sap_cloud_sdk/cbc/config.py @@ -0,0 +1,49 @@ +"""Configuration object for the CBC (Central Business Configuration) module.""" + +from __future__ import annotations + +import ssl +from collections.abc import Callable +from dataclasses import dataclass + + +@dataclass +class CBCConfig: + """Overrides for :func:`~sap_cloud_sdk.cbc.client.create_client`. + + Every field is optional. Left unset, :func:`create_client` infers the value + from the SAP application-platform provisioning conventions (runtime context, + tenant-mapping Fragment, Destination Service certificate). A value set here + wins over its ``CLOUD_SDK_CBC_*`` env var, which in turn falls back to the + platform default. + + The first three fields replace a whole default resolver; the last three steer + the default certificate / fragment lookup. + + Attributes: + base_url: CBC service base URL, as a string or a per-request callable. + Overrides the tenant-mapping Fragment lookup. + app_tenant_id: Application tenant id, as a string or a per-request + callable. Overrides reading ``APP_TENANT_ID`` from the runtime context. + ssl_context: Callable returning a freshly-built :class:`ssl.SSLContext` + with mTLS loaded. Overrides loading the certificate from the + Destination Service. ``None`` falls back to the default loader; use + this to inject a pre-built context (startup injection or tests), or a + plain-HTTP / local-mock path by returning an unauthenticated context. + destination_instance: The ``instance`` passed to the destination + ``create_fragment_client`` / ``create_certificate_client`` (used for + secret resolution in cloud mode). Falls back to the + ``CLOUD_SDK_CBC_DESTINATION_INSTANCE`` env var, else ``"default"``. + cbc_cert_name: Name of the app's mTLS certificate. Falls back to the + ``CLOUD_SDK_CBC_CERTIFICATE_NAME`` env var, else derived from the + platform landscape (``APPFND_CONHOS_LANDSCAPE``). + p12_password: Certificate keystore password. Falls back to the + ``CLOUD_SDK_CBC_P12_PASSWORD`` env var, else ``None``. + """ + + base_url: str | Callable[[], str] | None = None + app_tenant_id: str | Callable[[], str] | None = None + ssl_context: Callable[[], ssl.SSLContext] | None = None + destination_instance: str | None = None + cbc_cert_name: str | None = None + p12_password: bytes | None = None diff --git a/src/sap_cloud_sdk/cbc/exceptions.py b/src/sap_cloud_sdk/cbc/exceptions.py new file mode 100644 index 00000000..30b552ec --- /dev/null +++ b/src/sap_cloud_sdk/cbc/exceptions.py @@ -0,0 +1,89 @@ +"""Exception classes for the CBC (Central Business Configuration) module.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +class CBCError(Exception): + """Base exception for all CBC module errors.""" + + pass + + +class CBCConfigError(CBCError): + """Raised when CBC configuration is missing or unusable. + + Raised when no configuration can be resolved from the environment, or when + configuration is partially set (e.g. only one of the cert/key env vars is + provided, or a path env var points to a non-existent file). + """ + + pass + + +@dataclass(frozen=True) +class HttpContext: + """Context attached to HTTP exceptions. + + Attributes: + status_code: HTTP status code, or ``-1`` for network-level failures. + request_method: HTTP verb (GET, POST, PATCH, …). + request_url: Full request URL. + """ + + status_code: int + request_method: str + request_url: str + + +class CBCHttpError(CBCError): + """Raised for HTTP errors communicating with the CBC service. + + Attributes: + code: Application-level error code from the CBC API response, if available. + http_context: Request/response context. + """ + + def __init__( + self, + message: str, + code: str | None = None, + http_context: HttpContext | None = None, + ) -> None: + super().__init__(message) + self.message = message + self.code = code + self.http_context = http_context + + def __str__(self) -> str: + parts = [super().__str__()] + if self.code: + parts.append(f"code={self.code}") + if self.http_context: + parts.append(f"http_context={self.http_context}") + return " | ".join(parts) + + +class CBCClientError(CBCHttpError): + """Raised for 4xx responses from the CBC API.""" + + pass + + +class CBCServerError(CBCHttpError): + """Raised for 5xx responses from the CBC API.""" + + pass + + +class CBCNetworkError(CBCError): + """Raised for network-level failures (DNS, connection refused, timeouts).""" + + def __init__( + self, + message: str, + http_context: HttpContext | None = None, + ) -> None: + super().__init__(message) + self.http_context = http_context diff --git a/src/sap_cloud_sdk/cbc/py.typed b/src/sap_cloud_sdk/cbc/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/src/sap_cloud_sdk/cbc/user-guide.md b/src/sap_cloud_sdk/cbc/user-guide.md new file mode 100644 index 00000000..ca5fdbd4 --- /dev/null +++ b/src/sap_cloud_sdk/cbc/user-guide.md @@ -0,0 +1,309 @@ +# SAP Cloud SDK — CBC module + +Typed Python client for reading tenant-specific business configuration from +SAP Central Business Configuration (CBC). + +## Concepts + +**Consumption version** — a snapshot of the business configuration for one app +tenant at a point in time. An app tenant usually has one active version. + +**Configuration object** — a logical grouping of related configuration, e.g. +`payment-config` or `agent-config`. Authored by the application or agent team +and shipped as part of a reference content package. + +**Entity** — one slice of configuration within a config object. A config +object has one or more related entities; each entity has a stable authored `id` +(e.g. `payment-mode`) and a JSON schema defining its data shape. + +**Entity data** — the configuration content for an entity. + +## Setup + +The recommended entry point is `create_client()` with no arguments. On the SAP +application platform it infers everything from provisioning: + +- the application tenant id and the subscriber subdomain come from the SDK-wide + runtime context (wire `bootstrap(app)` once at startup — see below); +- the CBC URL comes from the tenant-mapping Destination Fragment written during + provisioning; +- the mTLS certificate comes from the Destination Service. + +```python +from sap_cloud_sdk import cbc + +cbc_client = cbc.create_client() +config = cbc_client.get_configuration() +``` + +The mTLS context is resolved once when the client is built and re-invoked only if +a request fails the TLS handshake, so a rotated certificate is picked up +automatically on the next call. + +To override any individual piece — for local development, tests, or a bespoke +deployment — pass a `CBCConfig`. See [Overrides](#overrides). + +## Reading configuration + +### Fetch everything in one call + +```python +# latest version resolved automatically +config = cbc_client.get_configuration() + +# pin a specific version +config = cbc_client.get_configuration(consumption_version="a0392d4f-72a9-...") + +# pick from the list +versions = cbc_client.get_consumption_versions() +cv = versions.latest() # or versions.items[0], or your own selection logic +config = cbc_client.get_configuration(consumption_version=cv.version) +``` + +### ConfigurationData structure + +``` +ConfigurationData +├── consumption_version: str # e.g. "a0392d4f-72a9-..." +├── app_tenant_id: str +└── configuration_objects: list[ConfigurationObject] + ├── ConfigurationObject + │ ├── configuration_object_id: str # e.g. "payment-config" + │ └── entities: list[ConfigurationEntity] + │ └── ConfigurationEntity + │ ├── entity_id: str # e.g. "payment-mode" + │ └── data: EntityData # .as_list() or .as_object() + └── ConfigurationObject + ├── configuration_object_id: str # e.g. "agent-config" + └── entities: list[ConfigurationEntity] + └── ConfigurationEntity + ├── entity_id: str # e.g. "contact" + └── data: EntityData # .as_list() or .as_object() +``` + +### Iterate all config objects and entities + +```python +for co in config.configuration_objects: + for ed in co.entities: + print(f"{co.configuration_object_id}/{ed.entity_id}") +``` + +### Look up a specific entity + +```python +# From already-fetched ConfigurationData +modes = config.get_entity_data("payment-config", "payment-mode") # EntityData | None +if modes: + for row in modes.as_list(): + print(row["paymentModeCode"], row["name"]) +``` + +If you need all entities for one config object, use `get_configuration_object` to get the +`ConfigurationObject` and iterate its `entities` list. + +`modes.as_list()` returns `list[dict]` and raises `ValueError` if the data is not a list. +`modes.as_object()` returns `dict` and raises `ValueError` if the data is not a dict. + +If you don't know the shape in advance, check first with `modes.is_list()` / +`modes.is_object()`, or call `modes.value()` to get the raw `list[dict] | dict` +without any shape assertion: + +```python +if modes.is_list(): + rows = modes.as_list() +else: + settings = modes.as_object() +``` + +```python +# Unmarshal into your own class +modes_list = [PaymentMode(**row) for row in modes.as_list()] +policy = PolicyConfig(**policy_entity.as_object()) +``` + +### Fetch a single entity directly + +When you only need one entity, use `get_entity_data` to avoid fetching all +configuration objects and their entities: + +```python +# One targeted HTTP call — no full config fetch +modes = cbc_client.get_entity_data("payment-config", "payment-mode") +for row in modes.as_list(): + print(row["paymentModeCode"], row["name"]) + +# Pin a specific version +modes = cbc_client.get_entity_data( + "payment-config", "payment-mode", consumption_version="a0392d4f-..." +) +``` + + +## Error handling + +| Exception | When | +|---|---| +| `CBCClientError` | 4xx from CBC (e.g. tenant not found) | +| `CBCServerError` | 5xx from CBC | +| `CBCNetworkError` | connection failure | +| `CBCConfigError` | the client cannot resolve the tenant mapping or certificate from provisioning | + +```python +from sap_cloud_sdk import cbc + +try: + config = cbc_client.get_configuration() +except cbc.CBCClientError as e: + print(e.code, e.message) # e.g. "NOT_FOUND", "Tenant unknown" +except cbc.CBCNetworkError: + ... # retry / circuit-break +``` + +## Platform setup + +On the SAP application platform `create_client()` needs no arguments. It reads +the application tenant id and the subscriber subdomain from the SDK-wide +**runtime context**, resolves the CBC URL from the tenant's Destination Fragment, +and loads the mTLS certificate from the Destination Service. + +The two artifacts it depends on follow the platform convention: + +- the app's own provider-level **Destination**, holding the mTLS certificate; +- the tenant-mapping **Fragment**, carrying the CBC URL (written by the platform + during tenant provisioning). + +The runtime context is populated once at startup via `bootstrap(app)`; CBC then +reads `APP_TENANT_ID` and `DWC_SUBDOMAIN` from it per request. You do not set any +CBC-specific context by hand. + +```python +from sap_cloud_sdk import bootstrap, cbc + +bootstrap(app) # wires the runtime context (IAS app_tid, dwc-subdomain, …) + +# once, at startup — the mTLS context is built here +cbc_client = cbc.create_client() + +# per request: nothing CBC-specific to set — the runtime context already carries +# the app tenant id and subdomain for the current request. +config = cbc_client.get_configuration() +``` + +The CBC URL comes from the `cbcUrl` property of the tenant-mapping fragment. The +client fetches the `CBC_TenantMapping_` fragment for the subscriber's +subdomain directly by name and uses its `cbcUrl` verbatim. The certificate is the +app's own provider-level mTLS certificate, fetched from the Destination Service. + +**Performance note.** The fragment lookup (Destination Service HTTP call) runs +on every `get_configuration()` call, because the CBC URL is resolved fresh each +time. For high-throughput agents making frequent CBC calls, consider caching the +resolved URL at the application layer (e.g. per tenant, invalidated on +`CBCClientError` with a not-found code). + +**Certificate rotation.** The mTLS certificate is loaded from the Destination +Service and reloaded automatically when a request fails the TLS handshake (as +happens once a certificate has rotated or expired): the client rebuilds its +mTLS context from a freshly-fetched certificate and retries the request once. A +long-lived `create_client()` singleton therefore recovers from rotation on its +own — no need to recreate it. + +## Multi-tenancy + +- **Supported:** Yes +- **Authentication:** IAS (the `app_tid` claim) for tenant identity; provider-level + mTLS to CBC. The subscriber subdomain selects the tenant's CBC endpoint. +- **How to use:** Wire `bootstrap(app)` once at startup, then call + `create_client()` with no arguments. CBC is resolved per request from the + SDK-wide runtime context: the application tenant id (`APP_TENANT_ID`, the IAS + `app_tid` claim) is sent as the `appTenantId` query parameter, and the + subscriber subdomain (`DWC_SUBDOMAIN`) selects the tenant-mapping Destination + Fragment (`CBC_TenantMapping_`, fetched with + `AccessStrategy.SUBSCRIBER_ONLY`) whose `cbcUrl` is the per-tenant base URL. A + single client instance is safe to share across tenants and threads — nothing + tenant-specific is held on the client; each call reads the current request's + tenant from the runtime context. For local development or pro-code callers, + override `base_url` / `app_tenant_id` on `CBCConfig` (see [Overrides](#overrides)). +- **Further reading:** [Platform setup](#platform-setup); `sap_cloud_sdk.core.runtime_context`. + +## Overrides + +`create_client` accepts an optional `CBCConfig`. Any field left unset is inferred +from provisioning; a field you set wins over its `CLOUD_SDK_CBC_*` env var, which +in turn falls back to the platform default. + +| Variable | Default | Description | +|---|---|---| +| `APPFND_CONHOS_LANDSCAPE` | (platform-provided) | Landscape used to derive the certificate name `sap-managed-runtime-ias-{landscape}.pem` | +| `CLOUD_SDK_CBC_CERTIFICATE_NAME` | landscape-derived | Explicit certificate name, overriding the landscape derivation | +| `CLOUD_SDK_CBC_DESTINATION_INSTANCE` | `default` | The `instance` passed to the destination `create_fragment_client` / `create_certificate_client` (used for secret resolution in cloud mode) | +| `CLOUD_SDK_CBC_P12_PASSWORD` | (none) | Password for the certificate keystore, if encrypted | + +```python +from sap_cloud_sdk import cbc + +# steer the default certificate lookup +cbc_client = cbc.create_client( + cbc.CBCConfig(cbc_cert_name="my-cert.pem"), +) +``` + +`CBCConfig` also exposes three override hooks that replace a whole default +resolver — `base_url`, `app_tenant_id` (each a string or a per-request callable), +and `ssl_context`. Set only the axis you want to control: + +```python +from sap_cloud_sdk import cbc + +# e.g. supply your own mTLS context from a vault, keep provisioning for the rest +client = cbc.create_client( + cbc.CBCConfig(ssl_context=lambda: my_ctx), +) +``` + +The `ssl_context` callable you pass still participates in the automatic +reload-on-TLS-failure rotation described above. + +> **Note.** The mTLS loader is the default **only** in pure provisioning mode. +> Once you override `base_url` or `app_tenant_id` (for example, to point at a +> local mock), an unset `ssl_context` means plain HTTPS with the public CA bundle +> — set `ssl_context` explicitly if you still need a client certificate. + +```python +# local mock: plain HTTP, no certificate needed +client = cbc.create_client( + cbc.CBCConfig( + base_url="http://localhost:8001", + app_tenant_id="my-app-tenant", + ), +) +``` + +## Using a test double + +`create_client` returns a concrete `CBCClient`. For tests, inject a mock +`httpx.Client` when constructing one directly, or substitute your own stub object +that implements the three public methods: + +```python +from sap_cloud_sdk import cbc + + +class StubCBCClient: + def get_consumption_versions(self): ... + def get_configuration(self, consumption_version=None): + return cbc.ConfigurationData( + consumption_version="cv1", + app_tenant_id="app-t1", + configuration_objects=[], + ) + + def get_entity_data( + self, configuration_object_id, entity_id, consumption_version=None + ): ... + + +def test_my_service(): + service = MyService(cbc_client=StubCBCClient()) + ... +``` diff --git a/src/sap_cloud_sdk/core/telemetry/module.py b/src/sap_cloud_sdk/core/telemetry/module.py index c3ceb80a..2c64727d 100644 --- a/src/sap_cloud_sdk/core/telemetry/module.py +++ b/src/sap_cloud_sdk/core/telemetry/module.py @@ -13,6 +13,7 @@ class Module(str, Enum): AUDITLOG = "auditlog" AUDITLOG_NG = "auditlog_ng" BOOTSTRAP = "bootstrap" + CBC = "cbc" DATA_ANONYMIZATION = "data_anonymization" DESTINATION = "destination" DPI_NG = "dpi_ng" diff --git a/src/sap_cloud_sdk/core/telemetry/operation.py b/src/sap_cloud_sdk/core/telemetry/operation.py index 5832a4f7..4ae1893d 100644 --- a/src/sap_cloud_sdk/core/telemetry/operation.py +++ b/src/sap_cloud_sdk/core/telemetry/operation.py @@ -139,6 +139,11 @@ class Operation(str, Enum): ADMS_CONFIG_GET_APP_TENANT = "config_get_app_tenant" ADMS_CONFIG_DELETE_APP_TENANT = "config_delete_app_tenant" + # CBC Operations + CBC_GET_CONSUMPTION_VERSIONS = "get_consumption_versions" + CBC_GET_CONFIGURATION = "get_configuration" + CBC_GET_ENTITY_DATA = "get_entity_data" + # Bootstrap Operations BOOTSTRAP = "bootstrap" diff --git a/tests/cbc/__init__.py b/tests/cbc/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/cbc/integration/__init__.py b/tests/cbc/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/cbc/integration/cbc.feature b/tests/cbc/integration/cbc.feature new file mode 100644 index 00000000..fbbae6b4 --- /dev/null +++ b/tests/cbc/integration/cbc.feature @@ -0,0 +1,36 @@ +Feature: CBC (Central Business Configuration) Integration + + Background: + Given a configured CBC client and tenant context + + # ── Consumption Versions ───────────────────────────────────────────────────── + + Scenario: Fetch consumption versions returns at least one version + When I call get_consumption_versions + Then the result should contain at least one version + + Scenario: Latest consumption version is non-empty + When I call get_consumption_versions + Then the latest version should have a non-empty version string + + # ── Configuration ──────────────────────────────────────────────────────────── + + Scenario: Fetch full configuration returns ConfigurationData + When I call get_configuration + Then the result should be a ConfigurationData with a non-empty consumption_version + And the app_tenant_id should match the configured tenant + + Scenario: Full configuration contains at least one config object + When I call get_configuration + Then the result should contain at least one config object + + Scenario: Every entity within each config object has an entity_id and data + When I call get_configuration + Then every entity should have a non-empty entity_id + And every entity data should be accessible as a list or object + + # ── Single entity fetch ────────────────────────────────────────────────────── + + Scenario: Fetch a single entity directly returns EntityData + When I call get_entity_data for the first entity in the first config object + Then the result should be EntityData accessible as a list or object diff --git a/tests/cbc/integration/conftest.py b/tests/cbc/integration/conftest.py new file mode 100644 index 00000000..fc1294e4 --- /dev/null +++ b/tests/cbc/integration/conftest.py @@ -0,0 +1,92 @@ +"""Pytest fixtures for CBC integration tests. + +Tests exercise the real default provisioning path: ``cbc.create_client()`` with +no overrides, resolving the CBC URL from the tenant-mapping Destination Fragment +and the mTLS certificate from the Destination Service — the same way the +agentgateway integration tests drive ``create_client``. + +Configuration is read from the environment (optionally from a +``.env_integration_tests`` file at the repo root, loaded automatically): + + # Destination Service binding (standard SDK convention, instance "default") + CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID + CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTSECRET + CLOUD_SDK_CFG_DESTINATION_DEFAULT_URL + CLOUD_SDK_CFG_DESTINATION_DEFAULT_URI + CLOUD_SDK_CFG_DESTINATION_DEFAULT_IDENTITYZONE + + # CBC tenant to target (the client reads these from the runtime context) + CLOUD_SDK_CFG_CBC_DEFAULT_APP_TENANT_ID IAS app_tid of the subscriber tenant + CLOUD_SDK_CFG_CBC_DEFAULT_TENANT_SUBDOMAIN subscriber subdomain (selects the fragment) + + # landscape, used to derive the mTLS certificate name + CLOUD_SDK_CFG_CBC_DEFAULT_LANDSCAPE -> APPFND_CONHOS_LANDSCAPE + +When the binding or the tenant values are missing, integration tests skip. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest +from dotenv import load_dotenv + +from sap_cloud_sdk.cbc import CBCClient, create_client +from sap_cloud_sdk.core.runtime_context import APP_TENANT_ID, DWC_SUBDOMAIN +from sap_cloud_sdk.core.runtime_context._context import RuntimeContext, set_context + +ENV_APP_TENANT_ID = "CLOUD_SDK_CFG_CBC_DEFAULT_APP_TENANT_ID" +ENV_TENANT_SUBDOMAIN = "CLOUD_SDK_CFG_CBC_DEFAULT_TENANT_SUBDOMAIN" +ENV_LANDSCAPE = "CLOUD_SDK_CFG_CBC_DEFAULT_LANDSCAPE" +ENV_DESTINATION_CLIENT_ID = "CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID" + + +def _setup_cloud_mode() -> None: + """Load environment variables from .env_integration_tests if present.""" + env_file = Path(__file__).parents[3] / ".env_integration_tests" + if env_file.exists(): + load_dotenv(env_file) + + +def _require_app_tenant_id() -> str: + app_tid = os.environ.get(ENV_APP_TENANT_ID) + if not app_tid: + pytest.skip(f"CBC integration tests skipped — set {ENV_APP_TENANT_ID}.") + return app_tid + + +@pytest.fixture(scope="session") +def cbc_app_tenant_id() -> str: + _setup_cloud_mode() + return _require_app_tenant_id() + + +@pytest.fixture +def cbc_client() -> CBCClient: + """Build the client via the default provisioning path (no overrides). + + Seeds the SDK runtime context with the subscriber tenant id and subdomain — + what ``bootstrap(app)`` would populate from an incoming request — then lets + ``create_client()`` resolve the URL and mTLS certificate from the Destination + Service, exactly as it does in production. + """ + _setup_cloud_mode() + + if not os.environ.get(ENV_DESTINATION_CLIENT_ID): + pytest.skip( + f"CBC integration tests skipped — set the Destination binding " + f"({ENV_DESTINATION_CLIENT_ID} and friends)." + ) + app_tid = _require_app_tenant_id() + subdomain = os.environ.get(ENV_TENANT_SUBDOMAIN) + if not subdomain: + pytest.skip(f"CBC integration tests skipped — set {ENV_TENANT_SUBDOMAIN}.") + + landscape = os.environ.get(ENV_LANDSCAPE) + if landscape: + os.environ.setdefault("APPFND_CONHOS_LANDSCAPE", landscape) + + set_context(RuntimeContext({APP_TENANT_ID: app_tid, DWC_SUBDOMAIN: subdomain})) + return create_client() diff --git a/tests/cbc/integration/test_e2e_bdd.py b/tests/cbc/integration/test_e2e_bdd.py new file mode 100644 index 00000000..99d40c7e --- /dev/null +++ b/tests/cbc/integration/test_e2e_bdd.py @@ -0,0 +1,173 @@ +"""BDD integration tests for the CBC (Central Business Configuration) module. + +Run against a real CBC server via the default provisioning path. Supply the +Destination Service binding and the target tenant (see conftest.py for the full +list), then:: + + CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID=... \\ + CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTSECRET=... \\ + CLOUD_SDK_CFG_DESTINATION_DEFAULT_URL=... \\ + CLOUD_SDK_CFG_DESTINATION_DEFAULT_URI=... \\ + CLOUD_SDK_CFG_DESTINATION_DEFAULT_IDENTITYZONE=... \\ + CLOUD_SDK_CFG_CBC_DEFAULT_APP_TENANT_ID=... \\ + CLOUD_SDK_CFG_CBC_DEFAULT_TENANT_SUBDOMAIN=... \\ + CLOUD_SDK_CFG_CBC_DEFAULT_LANDSCAPE=... \\ + pytest tests/cbc/integration +""" + +from __future__ import annotations + +import pytest +from pytest_bdd import given, scenario, then, when + +from sap_cloud_sdk.cbc import ConfigurationData, ConsumptionVersions, EntityData +from sap_cloud_sdk.cbc.client import CBCClient + +pytestmark = pytest.mark.integration + + +# -- Shared step state --------------------------------------------------------- + + +@pytest.fixture +def ctx() -> dict: + return {} + + +# -- Scenarios ----------------------------------------------------------------- + + +@scenario("cbc.feature", "Fetch consumption versions returns at least one version") +def test_consumption_versions_non_empty(): + pass + + +@scenario("cbc.feature", "Latest consumption version is non-empty") +def test_latest_version_non_empty(): + pass + + +@scenario("cbc.feature", "Fetch full configuration returns ConfigurationData") +def test_get_configuration_returns_configuration_data(): + pass + + +@scenario("cbc.feature", "Full configuration contains at least one config object") +def test_configuration_has_configuration_objects(): + pass + + +@scenario( + "cbc.feature", "Every entity within each config object has an entity_id and data" +) +def test_every_entity_has_id_and_data(): + pass + + +@scenario( + "cbc.feature", + "Fetch a single entity directly returns EntityData", +) +def test_get_entity_data_returns_entity_data(): + pass + + +# -- Steps --------------------------------------------------------------------- + + +@given("a configured CBC client and tenant context") +def cbc_context(cbc_client: CBCClient, cbc_app_tenant_id: str): + pass + + +@when("I call get_consumption_versions") +def call_get_consumption_versions(ctx: dict, cbc_client: CBCClient): + ctx["versions"] = cbc_client.get_consumption_versions() + + +@when("I call get_configuration") +def call_get_configuration(ctx: dict, cbc_client: CBCClient): + ctx["config"] = cbc_client.get_configuration() + + +@then("the result should contain at least one version") +def assert_versions_non_empty(ctx: dict): + versions: ConsumptionVersions = ctx["versions"] + assert len(versions.items) >= 1 + + +@then("the latest version should have a non-empty version string") +def assert_latest_version_non_empty(ctx: dict): + versions: ConsumptionVersions = ctx["versions"] + latest = versions.latest() + assert latest is not None + assert latest.version + + +@then("the result should be a ConfigurationData with a non-empty consumption_version") +def assert_configuration_data_type(ctx: dict): + config: ConfigurationData = ctx["config"] + assert isinstance(config, ConfigurationData) + assert config.consumption_version + + +@then("the app_tenant_id should match the configured tenant") +def assert_app_tenant_id(ctx: dict, cbc_app_tenant_id: str): + config: ConfigurationData = ctx["config"] + assert config.app_tenant_id == cbc_app_tenant_id + + +@then("the result should contain at least one config object") +def assert_configuration_objects_non_empty(ctx: dict): + config: ConfigurationData = ctx["config"] + assert len(config.configuration_objects) >= 1 + + +@then("every entity should have a non-empty entity_id") +def assert_entity_ids(ctx: dict): + config: ConfigurationData = ctx["config"] + for co in config.configuration_objects: + for ed in co.entities: + assert ed.entity_id, ( + f"entity_id missing in configuration_object={co.configuration_object_id!r}" + ) + + +@then("every entity data should be accessible as a list or object") +def assert_entity_data_accessible(ctx: dict): + config: ConfigurationData = ctx["config"] + for co in config.configuration_objects: + for ed in co.entities: + raw = ed.data + try: + result = raw.as_list() + assert result is not None + except ValueError: + result = raw.as_object() + assert result is not None + + +@when("I call get_entity_data for the first entity in the first config object") +def call_get_entity_data(ctx: dict, cbc_client: CBCClient): + config: ConfigurationData = cbc_client.get_configuration() + assert config.configuration_objects, ( + "no config objects — cannot test get_entity_data" + ) + co = config.configuration_objects[0] + assert co.entities, f"no entities in {co.configuration_object_id!r}" + entity_id = co.entities[0].entity_id + ctx["entity_data"] = cbc_client.get_entity_data( + co.configuration_object_id, entity_id + ) + + +@then("the result should be EntityData accessible as a list or object") +def assert_single_entity_data(ctx: dict): + data: EntityData = ctx["entity_data"] + assert isinstance(data, EntityData) + try: + result = data.as_list() + assert result is not None + except ValueError: + result = data.as_object() + assert result is not None diff --git a/tests/cbc/unit/__init__.py b/tests/cbc/unit/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/cbc/unit/test_client.py b/tests/cbc/unit/test_client.py new file mode 100644 index 00000000..e5b529c1 --- /dev/null +++ b/tests/cbc/unit/test_client.py @@ -0,0 +1,618 @@ +"""Unit tests for CBCClient and create_client.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any, cast +from unittest.mock import MagicMock + +import httpx +import json +import pytest +import ssl + +from sap_cloud_sdk.cbc.client import CBCClient, create_client, _is_tls_failure +from sap_cloud_sdk.cbc.config import CBCConfig +from sap_cloud_sdk.cbc.exceptions import ( + CBCClientError, + CBCConfigError, + CBCNetworkError, + CBCServerError, +) +from sap_cloud_sdk.cbc._models import ConfigurationData, EntityData + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _app_tid(value: str = "app-tenant") -> Callable[[], str]: + return lambda: value + + +def _mock_response( + status_code: int = 200, + json_body: Any = None, + content: bytes | None = None, +) -> httpx.Response: + if content is not None: + return httpx.Response(status_code=status_code, content=content) + body = json.dumps(json_body or {}).encode() + return httpx.Response( + status_code=status_code, + content=body, + headers={"content-type": "application/json"}, + ) + + +def _make_client( + base_url: str = "https://cbc.example.ondemand.com", + app_tenant_id: Callable[[], str] | None = None, +) -> tuple[CBCClient, MagicMock]: + mock_http = MagicMock(spec=httpx.Client) + client = CBCClient( + base_url=lambda: base_url, + app_tenant_id=app_tenant_id or _app_tid(), + http_client=mock_http, + ) + return client, mock_http + + +# --------------------------------------------------------------------------- +# CBCClient — app_tenant_id callable +# --------------------------------------------------------------------------- + + +class TestAppTenantIdCallable: + def test_app_tenant_id_callable_on_each_call_invokes_callable(self): + call_count = 0 + + def app_tenant_id_fn() -> str: + nonlocal call_count + call_count += 1 + return "app-tenant" + + mock_http = MagicMock(spec=httpx.Client) + mock_http.request.return_value = _mock_response( + json_body={"items": [{"version": "cv1"}]} + ) + client = CBCClient( + base_url=lambda: "https://cbc.example.ondemand.com", + app_tenant_id=app_tenant_id_fn, + http_client=mock_http, + ) + client.get_consumption_versions() + client.get_consumption_versions() + assert call_count == 2 + + +# --------------------------------------------------------------------------- +# CBCClient — URL building +# --------------------------------------------------------------------------- + + +class TestConfigurationsUrl: + def test_configurations_url_with_base_and_suffix_joins_correctly(self): + client, _ = _make_client("https://my-tenant.cbc.example.ondemand.com") + url = client._configurations_url( + "https://my-tenant.cbc.example.ondemand.com", "/consumptionVersions" + ) + assert url == ( + "https://my-tenant.cbc.example.ondemand.com" + "/configuration/v1/consumptionVersions" + ) + + +# --------------------------------------------------------------------------- +# CBCClient — get_consumption_versions +# --------------------------------------------------------------------------- + + +class TestGetConsumptionVersions: + def test_get_consumption_versions_with_valid_response_returns_parsed_versions(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={"items": [{"version": "cv1"}]} + ) + result = client.get_consumption_versions() + assert len(result.items) == 1 + assert result.items[0].version == "cv1" + + def test_get_consumption_versions_on_404_raises_cbc_client_error(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + status_code=404, + content=b'{"error":{"code":"NOT_FOUND","message":"not found"}}', + ) + with pytest.raises(CBCClientError): + client.get_consumption_versions() + + def test_get_consumption_versions_on_500_raises_cbc_server_error(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response(status_code=500, content=b"") + with pytest.raises(CBCServerError): + client.get_consumption_versions() + + def test_get_consumption_versions_on_connection_failure_raises_cbc_network_error( + self, + ): + client, mock_http = _make_client() + mock_http.request.side_effect = httpx.ConnectError("refused") + with pytest.raises(CBCNetworkError): + client.get_consumption_versions() + + +# --------------------------------------------------------------------------- +# CBCClient — _get_configuration_objects +# --------------------------------------------------------------------------- + + +class TestGetConfigurationObjects: + def test_get_configuration_objects_with_multiple_objects_returns_grouped_configuration_objects( + self, + ): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={ + "items": [ + { + "configurationObjectId": "payment-config", + "entities": [{"entityId": "payment-mode"}], + }, + { + "configurationObjectId": "tax-config", + "entities": [ + {"entityId": "tax-category"}, + {"entityId": "tax-rate"}, + ], + }, + ] + } + ) + result = client._get_configuration_objects( + "https://cbc.example.ondemand.com", "app-tenant", "cv1" + ) + assert len(result.items) == 2 + assert result.items[0].configuration_object_id == "payment-config" + assert [e.entity_id for e in result.items[1].entities] == [ + "tax-category", + "tax-rate", + ] + + +# --------------------------------------------------------------------------- +# CBCClient — _fetch_entity_data +# --------------------------------------------------------------------------- + + +class TestFetchEntityData: + def test_fetch_entity_data_with_array_shape_returns_list_content(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={ + "contentShape": "ARRAY", + "content": {"items": [{"code": "STD"}], "adaptedKeys": []}, + } + ) + result = client._fetch_entity_data( + "https://cbc.example.ondemand.com", + "app-tenant", + "cv1", + "tax-config", + "tax-category", + ) + assert result.entity_id == "tax-category" + assert result.data.as_list() == [{"code": "STD"}] + + def test_fetch_entity_data_with_object_shape_returns_dict_content(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={ + "contentShape": "OBJECT", + "content": {"item": {"maxRetries": 3, "timeoutSeconds": 30}}, + } + ) + result = client._fetch_entity_data( + "https://cbc.example.ondemand.com", + "app-tenant", + "cv1", + "pmc-settings", + "globalSettings", + ) + assert result.entity_id == "globalSettings" + assert result.data.as_object() == {"maxRetries": 3, "timeoutSeconds": 30} + + def test_fetch_entity_data_with_absent_content_defaults_to_empty_list(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={"contentShape": "ARRAY"} + ) + result = client._fetch_entity_data( + "https://cbc.example.ondemand.com", + "app-tenant", + "cv1", + "policy-config", + "policy", + ) + assert result.data.as_list() == [] + + +# --------------------------------------------------------------------------- +# CBCClient — get_configuration +# --------------------------------------------------------------------------- + + +class TestGetConfiguration: + def test_get_configuration_without_version_resolves_latest_and_returns_configuration_data( + self, + ): + client, mock_http = _make_client() + versions_response = _mock_response(json_body={"items": [{"version": "v2"}]}) + configuration_objects_response = _mock_response(json_body={"items": []}) + mock_http.request.side_effect = [ + versions_response, + configuration_objects_response, + ] + + result = client.get_configuration() + assert isinstance(result, ConfigurationData) + assert result.consumption_version == "v2" + assert result.configuration_objects == [] + + def test_get_configuration_when_no_versions_exist_raises_cbc_client_error(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response(json_body={"items": []}) + with pytest.raises(CBCClientError, match="no consumption version"): + client.get_configuration() + + def test_get_configuration_with_explicit_version_skips_version_resolution(self): + client, mock_http = _make_client() + configuration_objects_response = _mock_response( + json_body={ + "items": [ + { + "configurationObjectId": "payment-config", + "entities": [{"entityId": "payment-mode"}], + } + ] + } + ) + data_response = _mock_response( + json_body={ + "contentShape": "ARRAY", + "content": {"items": [{"k": "v"}]}, + } + ) + mock_http.request.side_effect = [configuration_objects_response, data_response] + + result = client.get_configuration(consumption_version="cv1") + assert len(result.configuration_objects) == 1 + assert ( + result.configuration_objects[0].configuration_object_id == "payment-config" + ) + assert len(result.configuration_objects[0].entities) == 1 + assert result.configuration_objects[0].entities[0].entity_id == "payment-mode" + + def test_get_configuration_with_multiple_entities_resolves_base_url_once(self): + """One get_configuration resolves base_url once, not per HTTP request.""" + call_count = 0 + + def base_url_fn() -> str: + nonlocal call_count + call_count += 1 + return "https://cbc.example.ondemand.com" + + mock_http = MagicMock(spec=httpx.Client) + client = CBCClient( + base_url=base_url_fn, + app_tenant_id=_app_tid(), + http_client=mock_http, + ) + configuration_objects_response = _mock_response( + json_body={ + "items": [ + { + "configurationObjectId": "tax-config", + "entities": [ + {"entityId": "tax-category"}, + {"entityId": "tax-rate"}, + ], + } + ] + } + ) + data_response = _mock_response( + json_body={"contentShape": "ARRAY", "content": {"items": []}} + ) + # config-objects + 2 entity-data requests = 3 HTTP calls, 1 base_url resolve + mock_http.request.side_effect = [ + configuration_objects_response, + data_response, + data_response, + ] + + client.get_configuration(consumption_version="cv1") + assert call_count == 1 + + +# --------------------------------------------------------------------------- +# CBCClient — get_entity_data +# --------------------------------------------------------------------------- + + +class TestGetEntityData: + def test_get_entity_data_without_version_resolves_latest_and_returns_entity_data( + self, + ): + client, mock_http = _make_client() + versions_response = _mock_response(json_body={"items": [{"version": "v1"}]}) + entity_response = _mock_response( + json_body={ + "contentShape": "ARRAY", + "content": {"items": [{"code": "STD"}], "adaptedKeys": []}, + } + ) + mock_http.request.side_effect = [versions_response, entity_response] + + result = client.get_entity_data("payment-config", "payment-mode") + assert isinstance(result, EntityData) + assert result.as_list() == [{"code": "STD"}] + + def test_get_entity_data_with_pinned_version_skips_version_resolution(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response( + json_body={ + "contentShape": "OBJECT", + "content": {"item": {"key": "value"}}, + } + ) + + result = client.get_entity_data( + "agent-config", "settings", consumption_version="v42" + ) + assert isinstance(result, EntityData) + assert result.as_object() == {"key": "value"} + # Only one HTTP call — no version resolution + assert mock_http.request.call_count == 1 + + def test_get_entity_data_when_no_versions_exist_raises_cbc_client_error(self): + client, mock_http = _make_client() + mock_http.request.return_value = _mock_response(json_body={"items": []}) + with pytest.raises(CBCClientError, match="no consumption version"): + client.get_entity_data("payment-config", "payment-mode") + + +# --------------------------------------------------------------------------- +# CBCClient — context manager +# --------------------------------------------------------------------------- + + +class TestCBCClientContextManager: + def test_context_manager_on_exit_closes_http_client(self): + client, mock_http = _make_client() + with client: + pass + mock_http.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# create_client — override path (CBCConfig axes) +# --------------------------------------------------------------------------- + + +class TestCreateClient: + def test_create_client_with_base_url_and_app_tenant_id_overrides_returns_client( + self, + ): + client = create_client( + CBCConfig( + base_url="http://localhost:8001", + app_tenant_id="app-t1", + ) + ) + assert isinstance(client, CBCClient) + # String overrides are wrapped into callables and resolved per request. + assert client._base_url() == "http://localhost:8001" + assert client._resolve_app_tenant_id() == "app-t1" + + def test_create_client_with_callable_overrides_are_invoked_per_request(self): + client = create_client( + CBCConfig( + base_url=lambda: "https://cbc.example.ondemand.com", + app_tenant_id=lambda: "app-t1", + ) + ) + assert isinstance(client, CBCClient) + assert client._base_url() == "https://cbc.example.ondemand.com" + assert client._resolve_app_tenant_id() == "app-t1" + + def test_create_client_with_ssl_context_override_passes_it_through(self): + ctx = ssl.create_default_context() + client = create_client( + CBCConfig( + base_url="https://cbc.example.ondemand.com", + app_tenant_id="app-t1", + ssl_context=lambda: ctx, + ) + ) + assert isinstance(client, CBCClient) + factory = client._ssl_factory + assert factory is not None + assert factory() is ctx + + +# --------------------------------------------------------------------------- +# _is_tls_failure +# --------------------------------------------------------------------------- + + +def _tls_read_error() -> httpx.ReadError: + """A ReadError wrapping an ssl.SSLError, as httpx surfaces an expired cert. + + Mirrors the empirically-observed shape: httpx.ReadError whose cause chain + reaches an ssl.SSLError (SSLV3_ALERT_CERTIFICATE_EXPIRED) a couple of levels + down. + """ + ssl_err = ssl.SSLError("[SSL: SSLV3_ALERT_CERTIFICATE_EXPIRED] certificate expired") + read_err = httpx.ReadError("TLS handshake failed") + read_err.__cause__ = ssl_err + return read_err + + +class TestIsTlsFailure: + def test_is_tls_failure_with_read_error_wrapping_ssl_error_returns_true(self): + assert _is_tls_failure(_tls_read_error()) is True + + def test_is_tls_failure_with_connect_error_returns_false(self): + assert _is_tls_failure(httpx.ConnectError("connection refused")) is False + + def test_is_tls_failure_with_non_ssl_os_error_returns_false(self): + err = httpx.ReadError("read failed") + err.__cause__ = OSError("broken pipe") + assert _is_tls_failure(err) is False + + +# --------------------------------------------------------------------------- +# CBCClient — mTLS certificate rotation (reactive rebuild) +# --------------------------------------------------------------------------- + + +class TestCertRotation: + def _rotating_client( + self, factory: Callable[[], ssl.SSLContext] + ) -> tuple[CBCClient, list[MagicMock]]: + """A CBCClient whose _build_http_client hands out fresh mock http clients. + + Each rebuild appends a new mock to the returned list, so a test can + assert how many times (and with what behaviour) the client was rebuilt. + The provided ``factory`` is wired as ``ssl_context`` so its invocation + count reflects each reactive rebuild (the initial mock below is injected + directly, bypassing the factory, as the real ``http_client`` seam does). + """ + built: list[MagicMock] = [] + + def build() -> httpx.Client: + factory() # exercise the ssl factory, same as the real _build_http_client + mock = MagicMock(spec=httpx.Client) + built.append(mock) + return mock + + initial = MagicMock(spec=httpx.Client) + built.append(initial) + client = CBCClient( + base_url=lambda: "https://cbc.example.ondemand.com", + app_tenant_id=_app_tid(), + http_client=initial, + ssl_context=factory, + ) + client._build_http_client = build # ty: ignore[invalid-assignment] + return client, built + + def test_cert_rotation_on_tls_failure_rebuilds_client_and_retries_successfully( + self, + ): + calls = {"factory": 0} + + def factory() -> ssl.SSLContext: + calls["factory"] += 1 + return ssl.create_default_context() + + client, built = self._rotating_client(factory) + ok = _mock_response(json_body={"items": [{"version": "cv1"}]}) + built[0].request.side_effect = _tls_read_error() + # the rebuilt client answers ok; wire it the moment it is created + orig_build = client._build_http_client + + def build_then_prime() -> httpx.Client: + mock = cast(MagicMock, orig_build()) + mock.request.return_value = ok + return mock + + client._build_http_client = build_then_prime # ty: ignore[invalid-assignment] + + result = client.get_consumption_versions() + + assert len(built) == 2 # initial + one rebuild + assert calls["factory"] == 1 # factory invoked once, on the rebuild + built[1].request.assert_called_once() + assert result.items[0].version == "cv1" + + def test_cert_rotation_on_second_tls_failure_after_rebuild_propagates_error(self): + def factory() -> ssl.SSLContext: + return ssl.create_default_context() + + client, built = self._rotating_client(factory) + built[0].request.side_effect = _tls_read_error() + orig_build = client._build_http_client + + def build_then_fail() -> httpx.Client: + mock = cast(MagicMock, orig_build()) + mock.request.side_effect = _tls_read_error() + return mock + + client._build_http_client = build_then_fail # ty: ignore[invalid-assignment] + + with pytest.raises(CBCNetworkError): + client.get_consumption_versions() + + # exactly one rebuild → built[0] + built[1]; no third client + assert len(built) == 2 + built[1].request.assert_called_once() + + def test_cert_rotation_on_non_tls_transport_error_does_not_rebuild_client(self): + def factory() -> ssl.SSLContext: + return ssl.create_default_context() + + client, built = self._rotating_client(factory) + built[0].request.side_effect = httpx.ConnectError("connection refused") + + with pytest.raises(CBCNetworkError): + client.get_consumption_versions() + + # no rebuild — still only the initial mock + assert len(built) == 1 + + def test_cert_rotation_without_ssl_factory_does_not_rebuild_client(self): + mock_http = MagicMock(spec=httpx.Client) + mock_http.request.side_effect = _tls_read_error() + client = CBCClient( + base_url=lambda: "https://cbc.example.ondemand.com", + app_tenant_id=_app_tid(), + http_client=mock_http, + ssl_context=None, + ) + with pytest.raises(CBCNetworkError): + client.get_consumption_versions() + # one attempt only, no rebuild path + mock_http.request.assert_called_once() + + def test_cert_rotation_when_factory_fails_propagates_error_and_retains_old_client( + self, + ): + calls = {"factory": 0} + + def factory() -> ssl.SSLContext: + calls["factory"] += 1 + # The initial client is injected via http_client (factory not called + # at construction), so the first invocation is the reactive rebuild — + # which fails, as a rotated-but-unfetchable cert would. + raise CBCConfigError("could not fetch the mTLS certificate") + + mock_http = MagicMock(spec=httpx.Client) + mock_http.request.side_effect = _tls_read_error() + client = CBCClient( + base_url=lambda: "https://cbc.example.ondemand.com", + app_tenant_id=_app_tid(), + http_client=mock_http, + ssl_context=factory, + ) + original = client._http_client + + with pytest.raises(CBCConfigError, match="mTLS certificate"): + client.get_consumption_versions() + + # factory invoked once on the failed rebuild; client left on the prior one + assert calls["factory"] == 1 + assert client._http_client is original + mock_http.request.assert_called_once() diff --git a/tests/cbc/unit/test_create_client.py b/tests/cbc/unit/test_create_client.py new file mode 100644 index 00000000..7c5c90a2 --- /dev/null +++ b/tests/cbc/unit/test_create_client.py @@ -0,0 +1,300 @@ +"""Unit tests for the default platform resolvers and create_client's wiring. + +These cover the behaviour that used to live in the platform adapter: resolving +``app_tenant_id`` / ``base_url`` from the SDK runtime context, loading the mTLS +certificate from the Destination Service, and the config → env → platform-default +precedence that ``create_client`` applies when a :class:`CBCConfig` axis is unset. +""" + +from __future__ import annotations + +import ssl +from unittest.mock import MagicMock, patch + +import pytest + +from sap_cloud_sdk.cbc.client import ( + CBC_FRAGMENT_PREFIX, + ENV_CERT_NAME, + ENV_DESTINATION_INSTANCE, + ENV_LANDSCAPE, + CBCClient, + _load_ssl_context, + _resolve_app_tenant_id, + _resolve_base_url, + create_client, +) +from sap_cloud_sdk.cbc.config import CBCConfig +from sap_cloud_sdk.cbc.exceptions import CBCConfigError +from sap_cloud_sdk.core.runtime_context import APP_TENANT_ID, DWC_SUBDOMAIN +from sap_cloud_sdk.core.runtime_context._context import RuntimeContext, set_context +from sap_cloud_sdk.destination import AccessStrategy + + +@pytest.fixture(autouse=True) +def _reset_runtime_context(): + """Each test starts from an empty runtime context.""" + set_context(RuntimeContext()) + yield + set_context(RuntimeContext()) + + +def _seed_context(app_tenant_id: str = "", subdomain: str = "") -> None: + values = {} + if app_tenant_id: + values[APP_TENANT_ID] = app_tenant_id + if subdomain: + values[DWC_SUBDOMAIN] = subdomain + set_context(RuntimeContext(values)) + + +# --------------------------------------------------------------------------- +# _resolve_app_tenant_id +# --------------------------------------------------------------------------- + + +class TestResolveAppTenantId: + def test_resolve_app_tenant_id_with_context_value_returns_it(self): + _seed_context(app_tenant_id="app-t1") + assert _resolve_app_tenant_id() == "app-t1" + + def test_resolve_app_tenant_id_with_empty_context_raises_cbc_config_error(self): + with pytest.raises(CBCConfigError, match="app tenant id"): + _resolve_app_tenant_id() + + +# --------------------------------------------------------------------------- +# _resolve_base_url +# --------------------------------------------------------------------------- + + +class TestResolveBaseUrl: + def test_resolve_base_url_with_matching_fragment_returns_cbc_url(self): + _seed_context(app_tenant_id="app-t1", subdomain="appfnd-subscriber") + match = MagicMock() + match.name = f"{CBC_FRAGMENT_PREFIX}appfnd-subscriber" + match.properties = { + "appTenantId": "app-t1", + "cbcUrl": "https://cbc.example.cloud.sap", + } + fake_client = MagicMock() + fake_client.get_subaccount_fragment.return_value = match + + with patch( + "sap_cloud_sdk.destination.create_fragment_client", + return_value=fake_client, + ): + url = _resolve_base_url("default") + + assert url == "https://cbc.example.cloud.sap" + fake_client.get_subaccount_fragment.assert_called_once_with( + f"{CBC_FRAGMENT_PREFIX}appfnd-subscriber", + access_strategy=AccessStrategy.SUBSCRIBER_ONLY, + tenant="appfnd-subscriber", + ) + + def test_resolve_base_url_with_empty_subdomain_raises_cbc_config_error(self): + _seed_context(app_tenant_id="app-t1") + with pytest.raises(CBCConfigError, match="subdomain"): + _resolve_base_url("default") + + def test_resolve_base_url_with_no_matching_fragment_raises_cbc_config_error(self): + _seed_context(app_tenant_id="app-t1", subdomain="appfnd-subscriber") + fake_client = MagicMock() + fake_client.get_subaccount_fragment.return_value = None + with patch( + "sap_cloud_sdk.destination.create_fragment_client", + return_value=fake_client, + ): + with pytest.raises(CBCConfigError, match="No CBC mapping fragment"): + _resolve_base_url("default") + + def test_resolve_base_url_with_fragment_missing_cbc_url_raises_cbc_config_error( + self, + ): + _seed_context(app_tenant_id="app-t1", subdomain="appfnd-subscriber") + match = MagicMock() + match.name = f"{CBC_FRAGMENT_PREFIX}appfnd-subscriber" + match.properties = {"appTenantId": "app-t1"} + fake_client = MagicMock() + fake_client.get_subaccount_fragment.return_value = match + with patch( + "sap_cloud_sdk.destination.create_fragment_client", + return_value=fake_client, + ): + with pytest.raises(CBCConfigError, match="no 'cbcUrl'"): + _resolve_base_url("default") + + def test_resolve_base_url_when_destination_raises_wraps_as_cbc_config_error(self): + from sap_cloud_sdk.destination.exceptions import DestinationOperationError + + _seed_context(app_tenant_id="app-t1", subdomain="appfnd-subscriber") + fake_client = MagicMock() + fake_client.get_subaccount_fragment.side_effect = DestinationOperationError( + "failed to get fragment: token error" + ) + with patch( + "sap_cloud_sdk.destination.create_fragment_client", + return_value=fake_client, + ): + with pytest.raises(CBCConfigError, match="Could not resolve the CBC URL"): + _resolve_base_url("default") + + +# --------------------------------------------------------------------------- +# _load_ssl_context +# --------------------------------------------------------------------------- + + +class TestLoadSslContext: + def test_load_ssl_context_with_valid_cert_returns_ssl_context(self): + ctx = ssl.create_default_context() + cert = MagicMock() + cert.content = "" + cert.name = "my-cert.pem" + fake_cert_client = MagicMock() + fake_cert_client.get_subaccount_certificate.return_value = cert + + with ( + patch( + "sap_cloud_sdk.destination.create_certificate_client", + return_value=fake_cert_client, + ), + patch( + "sap_cloud_sdk.destination._cert_loader._load_pem", + return_value=ctx, + ) as load_pem, + ): + result = _load_ssl_context("default", "my-cert.pem", b"secret") + + assert result is ctx + load_pem.assert_called_once_with("", b"secret", "my-cert.pem") + + def test_load_ssl_context_when_cert_not_found_raises_cbc_config_error(self): + fake_cert_client = MagicMock() + fake_cert_client.get_subaccount_certificate.return_value = None + with patch( + "sap_cloud_sdk.destination.create_certificate_client", + return_value=fake_cert_client, + ): + with pytest.raises(CBCConfigError, match="not found"): + _load_ssl_context("default", "missing.pem", None) + + def test_load_ssl_context_when_destination_raises_wraps_as_cbc_config_error(self): + from sap_cloud_sdk.destination.exceptions import DestinationOperationError + + fake_cert_client = MagicMock() + fake_cert_client.get_subaccount_certificate.side_effect = ( + DestinationOperationError("token error") + ) + with patch( + "sap_cloud_sdk.destination.create_certificate_client", + return_value=fake_cert_client, + ): + with pytest.raises( + CBCConfigError, match="Could not fetch the mTLS certificate" + ): + _load_ssl_context("default", "my-cert.pem", None) + + +# --------------------------------------------------------------------------- +# create_client — default wiring (infer from provisioning) + precedence +# --------------------------------------------------------------------------- + + +class TestCreateClientDefaults: + def test_default_wiring_with_landscape_env_builds_ssl_context_from_cert( + self, monkeypatch + ): + monkeypatch.setenv(ENV_LANDSCAPE, "cbc-fndtst-dev-eu12") + with patch( + "sap_cloud_sdk.cbc.client._load_ssl_context", + return_value=ssl.create_default_context(), + ) as load: + client = create_client() + assert isinstance(client, CBCClient) + # The factory is resolved once at construction to build the mTLS client. + load.assert_called_once() + instance_arg, cert_arg, _pw = load.call_args.args + assert instance_arg == "default" + assert cert_arg == "sap-managed-runtime-ias-cbc-fndtst-dev-eu12.pem" + # The factory is re-invokable (used again on a TLS failure to reload). + assert client._ssl_factory is not None + client._ssl_factory() + assert load.call_count == 2 + + def test_env_overrides_apply_instance_and_cert_name(self, monkeypatch): + monkeypatch.setenv(ENV_DESTINATION_INSTANCE, "cbc-instance") + monkeypatch.setenv(ENV_CERT_NAME, "my-cert.pem") + with patch( + "sap_cloud_sdk.cbc.client._load_ssl_context", + return_value=ssl.create_default_context(), + ) as load: + create_client() + instance_arg, cert_arg, _pw = load.call_args.args + assert instance_arg == "cbc-instance" + assert cert_arg == "my-cert.pem" + + def test_configuration_object_applies_instance_and_cert_name(self): + with patch( + "sap_cloud_sdk.cbc.client._load_ssl_context", + return_value=ssl.create_default_context(), + ) as load: + create_client( + CBCConfig( + destination_instance="cbc-instance", + cbc_cert_name="my-cert.pem", + ) + ) + instance_arg, cert_arg, _pw = load.call_args.args + assert instance_arg == "cbc-instance" + assert cert_arg == "my-cert.pem" + + def test_config_value_wins_over_env(self, monkeypatch): + monkeypatch.setenv(ENV_DESTINATION_INSTANCE, "from-env") + monkeypatch.setenv(ENV_CERT_NAME, "from-env.pem") + with patch( + "sap_cloud_sdk.cbc.client._load_ssl_context", + return_value=ssl.create_default_context(), + ) as load: + create_client( + CBCConfig( + destination_instance="from-config", + cbc_cert_name="from-config.pem", + ) + ) + instance_arg, cert_arg, _pw = load.call_args.args + assert instance_arg == "from-config" + assert cert_arg == "from-config.pem" + + def test_without_landscape_or_cert_name_raises_cbc_config_error(self, monkeypatch): + monkeypatch.delenv(ENV_LANDSCAPE, raising=False) + monkeypatch.delenv(ENV_CERT_NAME, raising=False) + with pytest.raises(CBCConfigError, match=ENV_LANDSCAPE): + create_client() + + def test_default_resolvers_read_runtime_context_at_request_time(self, monkeypatch): + """With no overrides, base_url + app_tenant_id come from the runtime context.""" + monkeypatch.setenv(ENV_CERT_NAME, "my-cert.pem") + with patch( + "sap_cloud_sdk.cbc.client._load_ssl_context", + return_value=ssl.create_default_context(), + ): + client = create_client() + assert isinstance(client, CBCClient) + + _seed_context(app_tenant_id="app-t1", subdomain="appfnd-subscriber") + assert client._resolve_app_tenant_id() == "app-t1" + + fragment = MagicMock() + fragment.name = f"{CBC_FRAGMENT_PREFIX}appfnd-subscriber" + fragment.properties = { + "appTenantId": "app-t1", + "cbcUrl": "https://cbc.example.cloud.sap", + } + fake_fc = MagicMock() + fake_fc.get_subaccount_fragment.return_value = fragment + with patch( + "sap_cloud_sdk.destination.create_fragment_client", return_value=fake_fc + ): + assert client._base_url() == "https://cbc.example.cloud.sap" diff --git a/tests/cbc/unit/test_models.py b/tests/cbc/unit/test_models.py new file mode 100644 index 00000000..5bfc9b90 --- /dev/null +++ b/tests/cbc/unit/test_models.py @@ -0,0 +1,187 @@ +"""Unit tests for CBC data models.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from sap_cloud_sdk.cbc._models import ( + ApiError, + ConfigurationData, + ConfigurationObject, + ConsumptionVersion, + ConsumptionVersions, + ConfigurationEntity, + EntityData, +) + + +# --------------------------------------------------------------------------- +# ConsumptionVersions.latest() +# --------------------------------------------------------------------------- + + +class TestConsumptionVersionsLatest: + def _version( + self, + version: str, + modified: datetime | None = None, + created: datetime | None = None, + ) -> ConsumptionVersion: + return ConsumptionVersion( + version=version, + modifiedDate=modified, + createdDate=created, + ) + + def test_latest_with_empty_list_returns_none(self): + assert ConsumptionVersions(items=[]).latest() is None + + def test_latest_with_modified_dates_returns_most_recently_modified(self): + t1 = datetime(2024, 1, 1, tzinfo=timezone.utc) + t2 = datetime(2024, 6, 1, tzinfo=timezone.utc) + v = ConsumptionVersions( + items=[ + self._version("v1", modified=t1), + self._version("v2", modified=t2), + ] + ) + result = v.latest() + assert result is not None + assert result.version == "v2" + + def test_latest_without_modified_dates_returns_most_recently_created(self): + t1 = datetime(2024, 1, 1, tzinfo=timezone.utc) + t2 = datetime(2024, 6, 1, tzinfo=timezone.utc) + v = ConsumptionVersions( + items=[ + self._version("v1", created=t1), + self._version("v2", created=t2), + ] + ) + result = v.latest() + assert result is not None + assert result.version == "v2" + + def test_latest_without_any_dates_returns_last_item(self): + v = ConsumptionVersions(items=[self._version("v1"), self._version("v2")]) + result = v.latest() + assert result is not None + assert result.version == "v2" + + +# --------------------------------------------------------------------------- +# EntityData +# --------------------------------------------------------------------------- + + +class TestEntityData: + def test_as_list_with_list_content_returns_list(self): + ec = EntityData([{"k": "v"}]) + assert ec.as_list() == [{"k": "v"}] + + def test_as_list_with_dict_content_raises_value_error(self): + ec = EntityData({"k": "v"}) + with pytest.raises(ValueError, match="as_object"): + ec.as_list() + + def test_as_object_with_dict_content_returns_dict(self): + ec = EntityData({"k": "v"}) + assert ec.as_object() == {"k": "v"} + + def test_as_object_with_list_content_raises_value_error(self): + ec = EntityData([{"k": "v"}]) + with pytest.raises(ValueError, match="as_list"): + ec.as_object() + + def test_is_list_with_list_content_returns_true_and_is_object_returns_false(self): + ec = EntityData([{"k": "v"}]) + assert ec.is_list() is True + assert ec.is_object() is False + + def test_is_object_with_dict_content_returns_true_and_is_list_returns_false(self): + ec = EntityData({"k": "v"}) + assert ec.is_object() is True + assert ec.is_list() is False + + def test_value_with_any_content_returns_raw_without_shape_assertion(self): + assert EntityData([{"k": "v"}]).value() == [{"k": "v"}] + assert EntityData({"k": "v"}).value() == {"k": "v"} + + +# --------------------------------------------------------------------------- +# ConfigurationData helpers +# --------------------------------------------------------------------------- + + +class TestConfigurationData: + def _entity_data(self, entity_id: str) -> ConfigurationEntity: + return ConfigurationEntity( + entity_id=entity_id, data=EntityData([{"id": entity_id}]) + ) + + def _configuration_object( + self, configuration_object_id: str, *entity_ids: str + ) -> ConfigurationObject: + return ConfigurationObject( + configuration_object_id=configuration_object_id, + entities=[self._entity_data(eid) for eid in entity_ids], + ) + + def _config(self, *configuration_objects: ConfigurationObject) -> ConfigurationData: + return ConfigurationData( + consumption_version="cv1", + app_tenant_id="app-t1", + configuration_objects=list(configuration_objects), + ) + + def test_get_configuration_object_with_matching_id_returns_configuration_object( + self, + ): + config = self._config( + self._configuration_object("ObjA", "E1"), + self._configuration_object("ObjB", "E2"), + ) + result = config.get_configuration_object("ObjA") + assert result is not None + assert result.configuration_object_id == "ObjA" + + def test_get_configuration_object_with_missing_id_returns_none(self): + config = self._config(self._configuration_object("ObjA", "E1")) + assert config.get_configuration_object("Missing") is None + + def test_get_entity_data_with_matching_ids_returns_correct_entity_data(self): + config = self._config( + self._configuration_object("ObjA", "E1", "E2"), + ) + result = config.get_entity_data("ObjA", "E2") + assert result is not None + assert isinstance(result, EntityData) + assert result.as_list() == [{"id": "E2"}] + + def test_get_entity_data_with_missing_entity_id_returns_none(self): + config = self._config(self._configuration_object("ObjA", "E1")) + assert config.get_entity_data("ObjA", "Missing") is None + + +# --------------------------------------------------------------------------- +# ApiError.from_response +# --------------------------------------------------------------------------- + + +class TestApiError: + def test_from_response_with_cbc_error_envelope_parses_code_and_message(self): + body = b'{"error":{"code":"NOT_FOUND","message":"Resource not found"}}' + err = ApiError.from_response(body) + assert err.code == "NOT_FOUND" + assert err.message == "Resource not found" + + def test_from_response_with_empty_body_falls_back_to_unknown_error(self): + err = ApiError.from_response(None) + assert err.code == "UNKNOWN_ERROR" + + def test_from_response_with_unparseable_body_falls_back_to_unknown_error(self): + err = ApiError.from_response(b"not json") + assert err.code == "UNKNOWN_ERROR" + assert "not json" in err.message diff --git a/tests/core/unit/telemetry/test_module.py b/tests/core/unit/telemetry/test_module.py index 6d38697a..77cb965a 100644 --- a/tests/core/unit/telemetry/test_module.py +++ b/tests/core/unit/telemetry/test_module.py @@ -58,8 +58,9 @@ 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) == 16 + assert len(all_modules) == 17 assert Module.ADMS in all_modules + assert Module.CBC in all_modules assert Module.AGENT_MEMORY in all_modules assert Module.AGENTGATEWAY in all_modules assert Module.AICORE in all_modules diff --git a/tests/core/unit/telemetry/test_operation.py b/tests/core/unit/telemetry/test_operation.py index aeb10a1a..cc42e94f 100644 --- a/tests/core/unit/telemetry/test_operation.py +++ b/tests/core/unit/telemetry/test_operation.py @@ -317,5 +317,6 @@ def test_operation_count(self): all_operations = list(Operation) # 3 auditlog + 12 destination + 10 certificate + 10 fragment + 8 objectstore # + 2 extensibility + 7 aicore + 23 dms + 7 agentgateway + 13 agent_memory - # + 5 data_anonymization + 52 adms + 6 print + 94 dpi_ng + 1 bootstrap + 3 output_management = 256 - assert len(all_operations) == 256 + # + 5 data_anonymization + 52 adms + 6 print + 94 dpi_ng + 1 bootstrap + 3 output_management + # + 3 cbc = 259 + assert len(all_operations) == 259 diff --git a/uv.lock b/uv.lock index f6cd4a21..018d6de9 100644 --- a/uv.lock +++ b/uv.lock @@ -166,9 +166,9 @@ name = "aiologic" version = "0.17.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "sniffio" }, - { name = "typing-extensions" }, - { name = "wrapt" }, + { name = "sniffio", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "wrapt", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/f1/7a/d51f2fde1e8ae8a83431f8e97b7a71e9358cdb1d4d2ce6be387fa44d68de/aiologic-0.17.1.tar.gz", hash = "sha256:2e1b93b9e88ced318c2a63ad7b382688f40cbfe40e3d42258d49dc9c5aea179d", size = 252354, upload-time = "2026-06-27T20:41:33.25Z" } wheels = [ @@ -816,8 +816,8 @@ name = "culsans" version = "0.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "aiologic" }, - { name = "typing-extensions" }, + { name = "aiologic", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d9/e3/49afa1bc180e0d28008ec6bcdf82a4072d1c7a41032b5b759b60814ca4b0/culsans-0.11.0.tar.gz", hash = "sha256:0b43d0d05dce6106293d114c86e3fb4bfc63088cfe8ff08ed3fe36891447fe33", size = 107546, upload-time = "2025-12-31T23:15:38.196Z" } wheels = [