From 88034b29e5e44be6f00073567c7dcfcbfa0f45a8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Fri, 14 Aug 2026 14:52:01 +0100 Subject: [PATCH] Add Model Target Web API client support Add ``ModelTargetService`` and ``AsyncModelTargetService``, which cover the create, poll, download and delete lifecycle for standard and advanced Model Target datasets. The clients authenticate with OAuth2 client credentials and reuse the returned bearer token until it expires. Typed request structures describe the models, guide views and State-Based Model Target configuration which Vuforia accepts, and the Model Target error envelope is parsed into public exceptions. Closes #3119. Co-Authored-By: Claude Opus 5 (1M context) --- conftest.py | 10 + docs/source/api-reference.rst | 12 + docs/source/exceptions.rst | 9 + docs/source/index.rst | 76 ++ newsfragments/3119.change.rst | 2 + pyproject.toml | 21 + spelling_private_dict.txt | 3 + src/vws/__init__.py | 4 + src/vws/_model_targets.py | 323 +++++++ src/vws/async_model_target_service.py | 381 +++++++++ src/vws/exceptions/model_target_exceptions.py | 210 +++++ src/vws/model_target_datasets.py | 152 ++++ src/vws/model_target_service.py | 366 ++++++++ src/vws/reports.py | 132 +++ tests/conftest.py | 69 ++ tests/test_async_model_targets.py | 409 +++++++++ tests/test_model_targets.py | 802 ++++++++++++++++++ 17 files changed, 2981 insertions(+) create mode 100644 newsfragments/3119.change.rst create mode 100644 src/vws/_model_targets.py create mode 100644 src/vws/async_model_target_service.py create mode 100644 src/vws/exceptions/model_target_exceptions.py create mode 100644 src/vws/model_target_datasets.py create mode 100644 src/vws/model_target_service.py create mode 100644 tests/test_async_model_targets.py create mode 100644 tests/test_model_targets.py diff --git a/conftest.py b/conftest.py index 1a2445a7c..e4c927947 100644 --- a/conftest.py +++ b/conftest.py @@ -61,6 +61,16 @@ def fixture_mock_vws( monkeypatch.setenv(name="VWS_CLIENT_ACCESS_KEY", value=client_access_key) monkeypatch.setenv(name="VWS_CLIENT_SECRET_KEY", value=client_secret_key) monkeypatch.setenv(name="VWS_DATABASE_ID", value=database_id) + # The mock accepts one hard-coded pair of Model Target Web API OAuth2 + # credentials, which it does not expose. + monkeypatch.setenv( + name="VWS_MODEL_TARGET_CLIENT_ID", + value="client-id", + ) + monkeypatch.setenv( + name="VWS_MODEL_TARGET_CLIENT_SECRET", + value="client-secret", + ) # We use a low processing time so that tests run quickly. with MockVWS(processing_time_seconds=0.2) as mock: mock.add_cloud_database(cloud_database=database) diff --git a/docs/source/api-reference.rst b/docs/source/api-reference.rst index c2765245d..f43793ce0 100644 --- a/docs/source/api-reference.rst +++ b/docs/source/api-reference.rst @@ -17,6 +17,18 @@ API Reference :undoc-members: :members: +.. automodule:: vws.model_target_service + :undoc-members: + :members: + +.. automodule:: vws.async_model_target_service + :undoc-members: + :members: + +.. automodule:: vws.model_target_datasets + :undoc-members: + :members: + .. automodule:: vws.reports :undoc-members: :members: diff --git a/docs/source/exceptions.rst b/docs/source/exceptions.rst index f48730bfd..baea7b495 100644 --- a/docs/source/exceptions.rst +++ b/docs/source/exceptions.rst @@ -28,6 +28,15 @@ CloudRecoService exceptions :inherited-members: Exception :exclude-members: errno, filename, filename2, strerror +ModelTargetService exceptions +----------------------------- + +.. automodule:: vws.exceptions.model_target_exceptions + :members: + :show-inheritance: + :inherited-members: Exception + :exclude-members: errno, filename, filename2, strerror + Custom exceptions ----------------- diff --git a/docs/source/index.rst b/docs/source/index.rst index ab7bfb0a7..c8b5bdaf6 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -110,6 +110,82 @@ The report is generated in the background, and the URL it is served from expires # This database has no targets, so nothing has been recognized. assert not reco_counts_by_target_id +Model Targets +------------- + +Vuforia generates Model Target datasets from CAD models. +This uses OAuth2 client credentials, which are separate from the VWS server keys. + +Dataset generation happens in the background, and the generated dataset is downloaded as a zip file. + +.. clear-namespace + +.. code-block:: python + + """Generate a Model Target dataset and download it.""" + + import os + + from vws import ModelTargetService + from vws.model_target_datasets import ( + CadDataFormat, + GuideViewPosition, + ModelTargetDatasetType, + ModelTargetModel, + ModelTargetView, + ) + from vws.reports import ModelTargetDatasetStatuses + + client_id = os.environ["VWS_MODEL_TARGET_CLIENT_ID"] + client_secret = os.environ["VWS_MODEL_TARGET_CLIENT_SECRET"] + + model_target_client = ModelTargetService( + client_id=client_id, + client_secret=client_secret, + ) + + model = ModelTargetModel( + name="my_model", + cad_data_url="https://example.com/my_model.zip", + cad_data_format=CadDataFormat.ZIP, + views=[ + ModelTargetView( + name="front", + guide_view_position=GuideViewPosition( + rotation=[0.0, 0.0, 0.0, 1.0], + translation=[0.0, 0.0, 1.0], + ), + ), + ], + ) + + dataset_uuid = model_target_client.create_dataset( + name="my_dataset", + target_sdk="11.0", + models=[model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = model_target_client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.DONE + + dataset = model_target_client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + # The dataset is a zip file. + assert dataset.startswith(b"PK") + + model_target_client.delete_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + Testing ------- diff --git a/newsfragments/3119.change.rst b/newsfragments/3119.change.rst new file mode 100644 index 000000000..5660e237e --- /dev/null +++ b/newsfragments/3119.change.rst @@ -0,0 +1,2 @@ +Add support for the Model Target Web API. +``ModelTargetService`` and ``AsyncModelTargetService`` create standard and advanced Model Target datasets, wait for them to be generated, download them and delete them. diff --git a/pyproject.toml b/pyproject.toml index 7c7d50d69..da92a9be8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -322,14 +322,26 @@ exclude = [ ".venv" ] # Ideally we would limit the paths to the source code where we want to ignore names, # but Vulture does not enable this. ignore_names = [ + # Model Target model option values which this library does not use + # itself, from vws.model_target_datasets + "ADAPTIVE", + "ALWAYS", + "AR_CONTROLLER", # Public API classes imported by users from vws.transports "AsyncHTTPXTransport", + "AUTO", # Sphinx "autoclass_content", "autoclass_content", "autodoc_member_order", + "CAR", "copybutton_exclude", + "DAE", + "DEFAULT", + "DYNAMIC", "extensions", + "FALSE", + "FBX", # pytest fixtures - we name fixtures like this for this purpose "fixture_*", "html_show_copyright", @@ -340,25 +352,34 @@ ignore_names = [ "html_title", "htmlhelp_basename", "HTTPXTransport", + "IGES", "intersphinx_mapping", "language", "linkcheck_ignore", "linkcheck_retries", + "LOW_FEATURE_OBJECTS", "master_doc", + "NEVER", "nitpick_ignore", "nitpicky", + "OBJ", "project_copyright", + "PVZ", "pygments_style", # pytest configuration "pytest_collect_file", "pytest_plugins", "rst_prolog", + "SCAN", "source_suffix", "spelling_word_list_filename", + "STATIC", + "STL", "templates_path", "towncrier_draft_autoversion_mode", "towncrier_draft_include_empty", "towncrier_draft_working_directory", + "VRML", "warning_is_error", ] diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index 74e7d3856..09386f1d4 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -11,6 +11,7 @@ LicenseCheckFailed MatchProcessing MaxNumResultsOutOfRange MetadataTooLarge +OAuth OopsAnErrorOccurredPossiblyBadName OopsAnErrorOccurredPossiblyBadNameError ProjectHasNoApiAccess @@ -33,6 +34,7 @@ args ascii async asyncio +balancer beartype bool boolean @@ -71,6 +73,7 @@ json keyring kib kwargs +lifecycle linters linting login diff --git a/src/vws/__init__.py b/src/vws/__init__.py index a181cb818..b4d39cc04 100644 --- a/src/vws/__init__.py +++ b/src/vws/__init__.py @@ -1,8 +1,10 @@ """A library for Vuforia Web Services.""" +from .async_model_target_service import AsyncModelTargetService from .async_query import AsyncCloudRecoService from .async_vumark_service import AsyncVuMarkService from .async_vws import AsyncVWS +from .model_target_service import ModelTargetService from .query import CloudRecoService from .vumark_service import VuMarkService from .vws import VWS @@ -10,8 +12,10 @@ __all__ = [ "VWS", "AsyncCloudRecoService", + "AsyncModelTargetService", "AsyncVWS", "AsyncVuMarkService", "CloudRecoService", + "ModelTargetService", "VuMarkService", ] diff --git a/src/vws/_model_targets.py b/src/vws/_model_targets.py new file mode 100644 index 000000000..c391342c2 --- /dev/null +++ b/src/vws/_model_targets.py @@ -0,0 +1,323 @@ +"""Internal helpers for the Vuforia Model Target Web API.""" + +import base64 +import json +from collections.abc import Sequence # noqa: TC003 +from http import HTTPStatus +from typing import Any + +from beartype import BeartypeConf, beartype + +from vws.exceptions.custom_exceptions import ServerError +from vws.exceptions.model_target_exceptions import ( + ModelTargetAuthenticationError, + ModelTargetDatasetNotDoneError, + ModelTargetError, + ModelTargetOAuth2Error, + ModelTargetValidationError, + UnknownModelTargetDatasetError, +) +from vws.exceptions.vws_exceptions import TooManyRequestsError +from vws.model_target_datasets import ( # noqa: TC001 + ModelTargetDatasetType, + ModelTargetModel, + ModelTargetView, +) +from vws.reports import ModelTargetDatasetStatusReport +from vws.response import Response # noqa: TC001 + +OAUTH2_TOKEN_PATH = "/oauth2/token" # noqa: S105 +OAUTH2_TOKEN_BODY = b"grant_type=client_credentials" +OAUTH2_TOKEN_CONTENT_TYPE = "application/x-www-form-urlencoded" # noqa: S105 +JSON_CONTENT_TYPE = "application/json" + +_DATASET_COLLECTION_PATHS = { + "standard": "/modeltargets/datasets", + "advanced": "/modeltargets/advancedDatasets", +} +_EXCEPTIONS_BY_STATUS_CODE: dict[int, type[ModelTargetError]] = { + HTTPStatus.BAD_REQUEST: ModelTargetValidationError, + HTTPStatus.UNAUTHORIZED: ModelTargetAuthenticationError, + HTTPStatus.NOT_FOUND: UnknownModelTargetDatasetError, + HTTPStatus.UNPROCESSABLE_ENTITY: ModelTargetDatasetNotDoneError, +} + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def oauth2_token_headers( + *, client_id: str, client_secret: str +) -> dict[str, str]: + """Get the headers for a request for an access token. + + Args: + client_id: A Model Target Web API client ID. + client_secret: A Model Target Web API client secret. + + Returns: + The headers to send with a token request. + """ + credentials = f"{client_id}:{client_secret}".encode() + encoded_credentials = base64.b64encode(s=credentials).decode( + encoding="ascii", + ) + return { + "Authorization": f"Basic {encoded_credentials}", + "Content-Type": OAUTH2_TOKEN_CONTENT_TYPE, + } + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def access_token_from_response(*, response: Response) -> tuple[str, float]: + """Get an access token and its lifetime from a token response. + + Args: + response: The response from Vuforia's token endpoint. + + Returns: + The access token, and the number of seconds until it expires. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + if response.status_code != HTTPStatus.OK: + raise ModelTargetOAuth2Error(response=response) + + response_data = dict(json.loads(s=response.text)) + return response_data["access_token"], float(response_data["expires_in"]) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_collection_path(*, dataset_type: ModelTargetDatasetType) -> str: + """Get the path of the endpoint for datasets of a given type. + + Args: + dataset_type: The kind of dataset to get the path for. + + Returns: + The path of the dataset collection endpoint. + """ + return _DATASET_COLLECTION_PATHS[dataset_type.value] + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_path( + *, + dataset_type: ModelTargetDatasetType, + dataset_uuid: str, +) -> str: + """Get the path of the endpoint for one dataset. + + Args: + dataset_type: The kind of dataset to get the path for. + dataset_uuid: The UUID of the dataset. + + Returns: + The path of the dataset endpoint. + """ + collection_path = dataset_collection_path(dataset_type=dataset_type) + return f"{collection_path}/{dataset_uuid}" + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_status_path( + *, + dataset_type: ModelTargetDatasetType, + dataset_uuid: str, +) -> str: + """Get the path of the status endpoint for one dataset. + + Args: + dataset_type: The kind of dataset to get the path for. + dataset_uuid: The UUID of the dataset. + + Returns: + The path of the dataset status endpoint. + """ + return ( + dataset_path(dataset_type=dataset_type, dataset_uuid=dataset_uuid) + + "/status" + ) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_download_path( + *, + dataset_type: ModelTargetDatasetType, + dataset_uuid: str, +) -> str: + """Get the path of the download endpoint for one dataset. + + Args: + dataset_type: The kind of dataset to get the path for. + dataset_uuid: The UUID of the dataset. + + Returns: + The path of the dataset download endpoint. + """ + return ( + dataset_path(dataset_type=dataset_type, dataset_uuid=dataset_uuid) + + "/dataset" + ) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def _view_dict(*, view: ModelTargetView) -> dict[str, Any]: + """Get the request representation of a guide view. + + Args: + view: The guide view to represent. + + Returns: + The guide view, as it is sent to Vuforia. + """ + view_dict: dict[str, Any] = { + "name": view.name, + "guideViewPosition": { + "rotation": list(view.guide_view_position.rotation), + "translation": list(view.guide_view_position.translation), + }, + } + if view.states is not None: + view_dict["states"] = list(view.states) + + return view_dict + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def _model_dict(*, model: ModelTargetModel) -> dict[str, Any]: + """Get the request representation of a model. + + Args: + model: The model to represent. + + Returns: + The model, as it is sent to Vuforia. + """ + model_dict: dict[str, Any] = {"name": model.name} + optional_values: dict[str, str | None] = { + "automaticColoring": model.automatic_coloring, + "cadDataBlob": model.cad_data_blob, + "cadDataFormat": model.cad_data_format, + "cadDataUrl": model.cad_data_url, + "motionHint": model.motion_hint, + "optimizeTrackingFor": model.optimize_tracking_for, + "realisticAppearance": model.realistic_appearance, + "simplify": model.simplify, + "stateBasedConfigurationJsonString": ( + model.state_based_configuration_json_string + ), + "trackingMode": model.tracking_mode, + } + for field_name, value in optional_values.items(): + if value is not None: + model_dict[field_name] = str(object=value) + + if model.views is not None: + model_dict["views"] = [_view_dict(view=view) for view in model.views] + + return model_dict + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_request_body( + *, + name: str, + target_sdk: str, + models: Sequence[ModelTargetModel], +) -> bytes: + """Get the request body for creating a Model Target dataset. + + Args: + name: The name of the dataset. + target_sdk: The Vuforia Engine version to generate the dataset + for. + models: The models to generate the dataset from. + + Returns: + The body of the request. + """ + request_dict = { + "models": [_model_dict(model=model) for model in models], + "name": name, + "targetSdk": target_sdk, + } + return json.dumps(obj=request_dict).encode(encoding="utf-8") + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def raise_for_error(*, response: Response) -> None: + """Raise an exception for an unsuccessful Model Target Web API + response. + + Args: + response: A response from the Model Target Web API. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.ModelTargetValidationError: + Vuforia rejected the dataset creation request. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetDatasetNotDoneError: + The dataset has not been generated. + ~vws.exceptions.model_target_exceptions.ModelTargetError: Vuforia + returned another error. + ~vws.exceptions.custom_exceptions.ServerError: There is an error + with Vuforia's servers. + ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is + rate limiting access. + """ + if ( + response.status_code == HTTPStatus.TOO_MANY_REQUESTS + ): # pragma: no cover + # The Vuforia API returns a 429 response with no JSON body. + raise TooManyRequestsError(response=response) + + if ( + response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR + ): # pragma: no cover + raise ServerError(response=response) + + if response.status_code < HTTPStatus.BAD_REQUEST: + return + + exception_type = _EXCEPTIONS_BY_STATUS_CODE.get( + response.status_code, + ModelTargetError, + ) + raise exception_type(response=response) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def dataset_uuid_from_response(*, response: Response) -> str: + """Get the UUID of a created dataset. + + Args: + response: A response to a dataset creation request. + + Returns: + The UUID of the created dataset. + """ + response_data = dict(json.loads(s=response.text)) + return str(object=response_data["uuid"]) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +def status_report_from_response( + *, + response: Response, +) -> ModelTargetDatasetStatusReport: + """Get a dataset status report from a status response. + + Args: + response: A response to a dataset status request. + + Returns: + The status of the dataset. + """ + response_data = dict(json.loads(s=response.text)) + return ModelTargetDatasetStatusReport.from_response_dict( + response_dict=response_data, + ) diff --git a/src/vws/async_model_target_service.py b/src/vws/async_model_target_service.py new file mode 100644 index 000000000..67a5e901c --- /dev/null +++ b/src/vws/async_model_target_service.py @@ -0,0 +1,381 @@ +"""Async interface to the Vuforia Model Target Web API.""" + +import asyncio +import time +from collections.abc import Sequence # noqa: TC003 +from http import HTTPMethod +from typing import Self + +from beartype import BeartypeConf, beartype + +from vws._model_targets import ( + JSON_CONTENT_TYPE, + OAUTH2_TOKEN_BODY, + OAUTH2_TOKEN_PATH, + access_token_from_response, + dataset_collection_path, + dataset_download_path, + dataset_path, + dataset_request_body, + dataset_status_path, + dataset_uuid_from_response, + oauth2_token_headers, + raise_for_error, + status_report_from_response, +) +from vws.exceptions.model_target_exceptions import ( + ModelTargetDatasetTimeoutError, +) +from vws.model_target_datasets import ( # noqa: TC001 + ModelTargetDatasetType, + ModelTargetModel, +) +from vws.reports import ( + ModelTargetDatasetStatuses, + ModelTargetDatasetStatusReport, +) +from vws.response import Response # noqa: TC001 +from vws.transports import AsyncHTTPXTransport, AsyncTransport + +_TOKEN_EXPIRY_MARGIN_SECONDS = 60.0 + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class AsyncModelTargetService: + """An async interface to the Vuforia Model Target Web API.""" + + def __init__( + self, + *, + client_id: str, + client_secret: str, + base_vws_url: str = "https://vws.vuforia.com", + request_timeout_seconds: float | tuple[float, float] = 30.0, + transport: AsyncTransport | None = None, + ) -> None: + """ + Args: + client_id: A Model Target Web API OAuth2 client + ID. + client_secret: A Model Target Web API OAuth2 + client secret. + base_vws_url: The base URL for the VWS API, which + also serves the Model Target Web API. + request_timeout_seconds: The timeout for each + HTTP request. This can be a float to set both + the connect and read timeouts, or a + (connect, read) tuple. + transport: The async HTTP transport to use for + requests. Defaults to + ``AsyncHTTPXTransport()``. + """ + self._client_id = client_id + self._client_secret = client_secret + self._base_vws_url = base_vws_url + self._request_timeout_seconds = request_timeout_seconds + self._transport = ( + transport if transport is not None else AsyncHTTPXTransport() + ) + self._access_token: str | None = None + self._access_token_expiry_time = 0.0 + + async def aclose(self) -> None: + """Close the underlying transport if it supports closing.""" + await self._transport.aclose() + + async def __aenter__(self) -> Self: + """Enter the async context manager.""" + return self + + async def __aexit__(self, *_args: object) -> None: + """Exit the async context manager and close the transport.""" + await self.aclose() + + async def get_access_token(self) -> str: + """Get an OAuth2 access token for the Model Target Web API. + + A token is requested only when the client has no token which is + still valid, so this can be called before each request. + + Returns: + A bearer token. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. For example, the + given client ID and client secret may not match a set of + Model Target Web API credentials. + """ + request_time = time.monotonic() + if ( + self._access_token is not None + and request_time < self._access_token_expiry_time + ): + return self._access_token + + response = await self._transport( + method=HTTPMethod.POST, + url=self._base_vws_url.rstrip("/") + OAUTH2_TOKEN_PATH, + headers=oauth2_token_headers( + client_id=self._client_id, + client_secret=self._client_secret, + ), + data=OAUTH2_TOKEN_BODY, + request_timeout=self._request_timeout_seconds, + ) + + access_token, expires_in_seconds = access_token_from_response( + response=response, + ) + self._access_token = access_token + self._access_token_expiry_time = ( + request_time + expires_in_seconds - _TOKEN_EXPIRY_MARGIN_SECONDS + ) + return access_token + + async def make_request( + self, + *, + method: str, + data: bytes, + request_path: str, + extra_headers: dict[str, str] | None = None, + ) -> Response: + """Make an authenticated request to the Model Target Web API. + + Args: + method: The HTTP method which will be used in + the request. + data: The request body which will be used in the + request. + request_path: The path to the endpoint which + will be used in the request. + extra_headers: Additional headers to include in + the request. + + Returns: + The response to the request. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetError: + Vuforia returned an error. + ~vws.exceptions.custom_exceptions.ServerError: + There is an error with Vuforia's servers. + ~vws.exceptions.vws_exceptions.TooManyRequestsError: + Vuforia is rate limiting access. + """ + access_token = await self.get_access_token() + headers = { + "Authorization": f"Bearer {access_token}", + **(extra_headers or {}), + } + + response = await self._transport( + method=method, + url=self._base_vws_url.rstrip("/") + request_path, + headers=headers, + data=data, + request_timeout=self._request_timeout_seconds, + ) + + raise_for_error(response=response) + return response + + async def create_dataset( + self, + *, + name: str, + target_sdk: str, + models: Sequence[ModelTargetModel], + dataset_type: ModelTargetDatasetType, + ) -> str: + """Start generating a Model Target dataset. + + Vuforia generates the dataset in the background, so it is not + available to download immediately. Use + :meth:`wait_for_dataset_generated` to wait for it. + + Args: + name: The name of the dataset. + target_sdk: The Vuforia Engine version to generate the dataset + for. + models: The models to generate the dataset from. A standard + dataset takes exactly one model. + dataset_type: Whether to create a standard or an advanced + dataset. + + Returns: + The UUID of the new dataset. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.ModelTargetValidationError: + Vuforia rejected the request. For example, a model may + give neither a CAD data URL nor a CAD data blob. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = await self.make_request( + method=HTTPMethod.POST, + data=dataset_request_body( + name=name, + target_sdk=target_sdk, + models=models, + ), + request_path=dataset_collection_path(dataset_type=dataset_type), + extra_headers={"Content-Type": JSON_CONTENT_TYPE}, + ) + + return dataset_uuid_from_response(response=response) + + async def get_dataset_status( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> ModelTargetDatasetStatusReport: + """Get the status of a Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to get the status of. + + Returns: + The status of the dataset. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = await self.make_request( + method=HTTPMethod.GET, + data=b"", + request_path=dataset_status_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) + + return status_report_from_response(response=response) + + async def wait_for_dataset_generated( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + seconds_between_requests: float = 0.2, + timeout_seconds: float = 60 * 5, + ) -> ModelTargetDatasetStatusReport: + """Wait for Vuforia to finish generating a Model Target dataset. + + A dataset which failed to generate is also finished, so the + returned report may have a + :attr:`~.ModelTargetDatasetStatusReport.status` of + ``FAILED``. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to wait for. + seconds_between_requests: The number of seconds to wait between + requests made while polling the dataset's status. + timeout_seconds: The maximum number of seconds to wait for the + dataset to be generated. + + Returns: + The status of the dataset once it is no longer processing. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetDatasetTimeoutError: + The dataset was not generated within ``timeout_seconds`` + seconds. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + """ + start_time = time.monotonic() + while True: + report = await self.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + if report.status != ModelTargetDatasetStatuses.PROCESSING: + return report + + elapsed_time = time.monotonic() - start_time + if elapsed_time > timeout_seconds: + raise ModelTargetDatasetTimeoutError + + await asyncio.sleep(delay=seconds_between_requests) + + async def download_dataset( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> bytes: + """Download a generated Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to download. + + Returns: + The dataset, as the bytes of a zip file. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetDatasetNotDoneError: + Vuforia has not generated the dataset. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = await self.make_request( + method=HTTPMethod.GET, + data=b"", + request_path=dataset_download_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) + + return response.content + + async def delete_dataset( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> None: + """Delete a Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to delete. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + await self.make_request( + method=HTTPMethod.DELETE, + data=b"", + request_path=dataset_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) diff --git a/src/vws/exceptions/model_target_exceptions.py b/src/vws/exceptions/model_target_exceptions.py new file mode 100644 index 000000000..8eaf09878 --- /dev/null +++ b/src/vws/exceptions/model_target_exceptions.py @@ -0,0 +1,210 @@ +"""Exceptions raised by the Vuforia Model Target Web API. + +See +https://developer.vuforia.com/library/vuforia-engine/web-api/model-target-web-api/. +""" + +import json +from typing import Any + +from beartype import beartype + +from vws.reports import ModelTargetGenerationDetail +from vws.response import Response # noqa: TC001 + + +@beartype +def _is_json_object(*, value: object) -> bool: + """Get whether a decoded JSON value is an object. + + Args: + value: A decoded JSON value. + + Returns: + Whether the value is a JSON object. + """ + return isinstance(value, dict) + + +@beartype +def _json_object(*, value: str) -> dict[str, Any]: + """Get a JSON object from a string. + + Args: + value: A string which may be a JSON object. + + Returns: + The JSON object, or an empty dictionary if the string is not a + JSON object. + """ + try: + loaded: Any = json.loads(s=value) + except json.JSONDecodeError: + return {} + + if not _is_json_object(value=loaded): + return {} + + json_object: dict[str, Any] = loaded + return json_object + + +@beartype +def _error_dict(*, response: Response) -> dict[str, Any]: + """Get the error object of a Model Target Web API error response. + + Args: + response: The response returned by Vuforia. + + Returns: + The error object, or an empty dictionary if the response has no + error object. Some errors, such as those given by the load + balancer in front of Vuforia, are not shaped like Model Target + Web API errors. + """ + body = _json_object(value=response.text) + if "error" not in body: + return {} + + error: Any = body["error"] + if not _is_json_object(value=error): + return {} + + error_dict: dict[str, Any] = error + return error_dict + + +@beartype +class ModelTargetError(Exception): + """Base class for Vuforia Model Target Web API exceptions.""" + + def __init__(self, response: Response) -> None: + """ + Args: + response: The response to a request to Vuforia. + """ + super().__init__(response.text) + self._response = response + + @property + def response(self) -> Response: + """The response returned by Vuforia which included this error.""" + return self._response + + @property + def code(self) -> str: + """The error code given by Vuforia, or an empty string.""" + error = _error_dict(response=self._response) + return str(object=error["code"]) if "code" in error else "" + + @property + def message(self) -> str: + """The error message given by Vuforia, or an empty string.""" + error = _error_dict(response=self._response) + return str(object=error["message"]) if "message" in error else "" + + @property + def target(self) -> str: + """The error target given by Vuforia, or an empty string.""" + error = _error_dict(response=self._response) + return str(object=error["target"]) if "target" in error else "" + + @property + def details(self) -> list[ModelTargetGenerationDetail]: + """The error details given by Vuforia. + + Vuforia gives one detail per validation problem it found with a + dataset creation request. + """ + error = _error_dict(response=self._response) + if "details" not in error: + return [] + + return [ + ModelTargetGenerationDetail( + code=detail["code"], + message=detail["message"], + ) + for detail in error["details"] + ] + + +@beartype +class ModelTargetAuthenticationError(ModelTargetError): + """Exception raised when a Model Target Web API request is not + authenticated. + + For example, the bearer token may be missing, malformed or expired. + """ + + +@beartype +class ModelTargetValidationError(ModelTargetError): + """Exception raised when Vuforia rejects a Model Target dataset + creation + request. + + See :attr:`~.ModelTargetError.details` for the problems which Vuforia + found. + """ + + +@beartype +class UnknownModelTargetDatasetError(ModelTargetError): + """Exception raised when no Model Target dataset matches a given UUID. + + Standard and advanced datasets are separate resources, so this is also + raised when the given UUID matches a dataset of the other type. + """ + + +@beartype +class ModelTargetDatasetNotDoneError(ModelTargetError): + """Exception raised when a Model Target dataset is downloaded before + Vuforia has generated it. + """ + + +@beartype +class ModelTargetOAuth2Error(Exception): + """Exception raised when Vuforia does not give an access token. + + For example, the given client ID and client secret may not match a set + of Model Target Web API credentials. + """ + + def __init__(self, response: Response) -> None: + """ + Args: + response: The response to a request to Vuforia's token + endpoint. + """ + super().__init__(response.text) + self._response = response + + @property + def response(self) -> Response: + """The response returned by Vuforia which included this error.""" + return self._response + + @property + def error(self) -> str: + """The OAuth2 error code, or an empty string.""" + body = _json_object(value=self._response.text) + return str(object=body["error"]) if "error" in body else "" + + @property + def error_description(self) -> str: + """The OAuth2 error description, or an empty string.""" + body = _json_object(value=self._response.text) + if "error_description" not in body: + return "" + + return str(object=body["error_description"]) + + +@beartype +class ModelTargetDatasetTimeoutError(Exception): + """Exception raised when waiting for a Model Target dataset to be + generated times out. + """ diff --git a/src/vws/model_target_datasets.py b/src/vws/model_target_datasets.py new file mode 100644 index 000000000..95a9eaa18 --- /dev/null +++ b/src/vws/model_target_datasets.py @@ -0,0 +1,152 @@ +"""Structures for describing Model Target datasets to create. + +See +https://developer.vuforia.com/library/vuforia-engine/web-api/model-target-web-api/. +""" + +from collections.abc import Sequence # noqa: TC003 +from dataclasses import dataclass +from enum import StrEnum, unique + +from beartype import BeartypeConf, beartype + + +@beartype +@unique +class ModelTargetDatasetType(StrEnum): + """The kinds of Model Target dataset which Vuforia generates. + + Standard and advanced datasets are separate resources, so a dataset + created as one type is not visible to requests for the other type. + """ + + STANDARD = "standard" + ADVANCED = "advanced" + + +@beartype +@unique +class AutomaticColoring(StrEnum): + """Options for a model's ``automaticColoring``.""" + + ALWAYS = "always" + AUTO = "auto" + NEVER = "never" + + +@beartype +@unique +class CadDataFormat(StrEnum): + """Options for a model's ``cadDataFormat``.""" + + DAE = "DAE" + FBX = "FBX" + GLB = "GLB" + IGES = "IGES" + OBJ = "OBJ" + PVZ = "PVZ" + STL = "STL" + VRML = "VRML" + ZIP = "ZIP" + + +@beartype +@unique +class MotionHint(StrEnum): + """Options for a model's ``motionHint``.""" + + ADAPTIVE = "adaptive" + DYNAMIC = "dynamic" + STATIC = "static" + + +@beartype +@unique +class OptimizeTrackingFor(StrEnum): + """Options for a model's ``optimizeTrackingFor``.""" + + AR_CONTROLLER = "ar_controller" + DEFAULT = "default" + LOW_FEATURE_OBJECTS = "low_feature_objects" + + +@beartype +@unique +class RealisticAppearance(StrEnum): + """Options for a model's ``realisticAppearance``. + + This is documented for advanced datasets only. + """ + + AUTO = "auto" + FALSE = "false" + TRUE = "true" + + +@beartype +@unique +class Simplify(StrEnum): + """Options for a model's ``simplify``.""" + + ALWAYS = "always" + AUTO = "auto" + NEVER = "never" + + +@beartype +@unique +class TrackingMode(StrEnum): + """Options for a model's ``trackingMode``.""" + + CAR = "car" + DEFAULT = "default" + SCAN = "scan" + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +@dataclass(frozen=True, kw_only=True) +class GuideViewPosition: + """The position of a guide view.""" + + rotation: Sequence[float] + translation: Sequence[float] + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetView: + """A guide view of a model.""" + + name: str + guide_view_position: GuideViewPosition + states: Sequence[str] | None = None + """The State-Based Model Target states which this view applies to. + + Every given state must be named by the model's + ``state_based_configuration_json_string``. + """ + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetModel: + """A model to generate a Model Target dataset from. + + One and only one of ``cad_data_url`` and ``cad_data_blob`` is + required. + """ + + name: str + cad_data_url: str | None = None + cad_data_blob: str | None = None + automatic_coloring: AutomaticColoring | None = None + cad_data_format: CadDataFormat | None = None + motion_hint: MotionHint | None = None + optimize_tracking_for: OptimizeTrackingFor | None = None + realistic_appearance: RealisticAppearance | None = None + """This is documented for advanced datasets only.""" + + simplify: Simplify | None = None + tracking_mode: TrackingMode | None = None + state_based_configuration_json_string: str | None = None + views: Sequence[ModelTargetView] | None = None diff --git a/src/vws/model_target_service.py b/src/vws/model_target_service.py new file mode 100644 index 000000000..70b116d0a --- /dev/null +++ b/src/vws/model_target_service.py @@ -0,0 +1,366 @@ +"""Interface to the Vuforia Model Target Web API.""" + +import time +from collections.abc import Sequence # noqa: TC003 +from http import HTTPMethod + +from beartype import BeartypeConf, beartype + +from vws._model_targets import ( + JSON_CONTENT_TYPE, + OAUTH2_TOKEN_BODY, + OAUTH2_TOKEN_PATH, + access_token_from_response, + dataset_collection_path, + dataset_download_path, + dataset_path, + dataset_request_body, + dataset_status_path, + dataset_uuid_from_response, + oauth2_token_headers, + raise_for_error, + status_report_from_response, +) +from vws.exceptions.model_target_exceptions import ( + ModelTargetDatasetTimeoutError, +) +from vws.model_target_datasets import ( # noqa: TC001 + ModelTargetDatasetType, + ModelTargetModel, +) +from vws.reports import ( + ModelTargetDatasetStatuses, + ModelTargetDatasetStatusReport, +) +from vws.response import Response # noqa: TC001 +from vws.transports import RequestsTransport, Transport + +_TOKEN_EXPIRY_MARGIN_SECONDS = 60.0 + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class ModelTargetService: + """An interface to the Vuforia Model Target Web API.""" + + def __init__( + self, + *, + client_id: str, + client_secret: str, + base_vws_url: str = "https://vws.vuforia.com", + request_timeout_seconds: float | tuple[float, float] = 30.0, + transport: Transport | None = None, + ) -> None: + """ + Args: + client_id: A Model Target Web API OAuth2 client + ID. + client_secret: A Model Target Web API OAuth2 + client secret. + base_vws_url: The base URL for the VWS API, which + also serves the Model Target Web API. + request_timeout_seconds: The timeout for each + HTTP request. This can be a float to set both + the connect and read timeouts, or a + (connect, read) tuple. + transport: The HTTP transport to use for + requests. Defaults to + ``RequestsTransport()``. + """ + self._client_id = client_id + self._client_secret = client_secret + self._base_vws_url = base_vws_url + self._request_timeout_seconds = request_timeout_seconds + self._transport = ( + transport if transport is not None else RequestsTransport() + ) + self._access_token: str | None = None + self._access_token_expiry_time = 0.0 + + def get_access_token(self) -> str: + """Get an OAuth2 access token for the Model Target Web API. + + A token is requested only when the client has no token which is + still valid, so this can be called before each request. + + Returns: + A bearer token. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. For example, the + given client ID and client secret may not match a set of + Model Target Web API credentials. + """ + request_time = time.monotonic() + if ( + self._access_token is not None + and request_time < self._access_token_expiry_time + ): + return self._access_token + + response = self._transport( + method=HTTPMethod.POST, + url=self._base_vws_url.rstrip("/") + OAUTH2_TOKEN_PATH, + headers=oauth2_token_headers( + client_id=self._client_id, + client_secret=self._client_secret, + ), + data=OAUTH2_TOKEN_BODY, + request_timeout=self._request_timeout_seconds, + ) + + access_token, expires_in_seconds = access_token_from_response( + response=response, + ) + self._access_token = access_token + self._access_token_expiry_time = ( + request_time + expires_in_seconds - _TOKEN_EXPIRY_MARGIN_SECONDS + ) + return access_token + + def make_request( + self, + *, + method: str, + data: bytes, + request_path: str, + extra_headers: dict[str, str] | None = None, + ) -> Response: + """Make an authenticated request to the Model Target Web API. + + Args: + method: The HTTP method which will be used in + the request. + data: The request body which will be used in the + request. + request_path: The path to the endpoint which + will be used in the request. + extra_headers: Additional headers to include in + the request. + + Returns: + The response to the request. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetError: + Vuforia returned an error. + ~vws.exceptions.custom_exceptions.ServerError: + There is an error with Vuforia's servers. + ~vws.exceptions.vws_exceptions.TooManyRequestsError: + Vuforia is rate limiting access. + """ + headers = { + "Authorization": f"Bearer {self.get_access_token()}", + **(extra_headers or {}), + } + + response = self._transport( + method=method, + url=self._base_vws_url.rstrip("/") + request_path, + headers=headers, + data=data, + request_timeout=self._request_timeout_seconds, + ) + + raise_for_error(response=response) + return response + + def create_dataset( + self, + *, + name: str, + target_sdk: str, + models: Sequence[ModelTargetModel], + dataset_type: ModelTargetDatasetType, + ) -> str: + """Start generating a Model Target dataset. + + Vuforia generates the dataset in the background, so it is not + available to download immediately. Use + :meth:`wait_for_dataset_generated` to wait for it. + + Args: + name: The name of the dataset. + target_sdk: The Vuforia Engine version to generate the dataset + for. + models: The models to generate the dataset from. A standard + dataset takes exactly one model. + dataset_type: Whether to create a standard or an advanced + dataset. + + Returns: + The UUID of the new dataset. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.ModelTargetValidationError: + Vuforia rejected the request. For example, a model may + give neither a CAD data URL nor a CAD data blob. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = self.make_request( + method=HTTPMethod.POST, + data=dataset_request_body( + name=name, + target_sdk=target_sdk, + models=models, + ), + request_path=dataset_collection_path(dataset_type=dataset_type), + extra_headers={"Content-Type": JSON_CONTENT_TYPE}, + ) + + return dataset_uuid_from_response(response=response) + + def get_dataset_status( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> ModelTargetDatasetStatusReport: + """Get the status of a Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to get the status of. + + Returns: + The status of the dataset. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = self.make_request( + method=HTTPMethod.GET, + data=b"", + request_path=dataset_status_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) + + return status_report_from_response(response=response) + + def wait_for_dataset_generated( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + seconds_between_requests: float = 0.2, + timeout_seconds: float = 60 * 5, + ) -> ModelTargetDatasetStatusReport: + """Wait for Vuforia to finish generating a Model Target dataset. + + A dataset which failed to generate is also finished, so the + returned report may have a + :attr:`~.ModelTargetDatasetStatusReport.status` of + ``FAILED``. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to wait for. + seconds_between_requests: The number of seconds to wait between + requests made while polling the dataset's status. + timeout_seconds: The maximum number of seconds to wait for the + dataset to be generated. + + Returns: + The status of the dataset once it is no longer processing. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetDatasetTimeoutError: + The dataset was not generated within ``timeout_seconds`` + seconds. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + """ + start_time = time.monotonic() + while True: + report = self.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + if report.status != ModelTargetDatasetStatuses.PROCESSING: + return report + + elapsed_time = time.monotonic() - start_time + if elapsed_time > timeout_seconds: + raise ModelTargetDatasetTimeoutError + + time.sleep(seconds_between_requests) + + def download_dataset( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> bytes: + """Download a generated Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to download. + + Returns: + The dataset, as the bytes of a zip file. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetDatasetNotDoneError: + Vuforia has not generated the dataset. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + response = self.make_request( + method=HTTPMethod.GET, + data=b"", + request_path=dataset_download_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) + + return response.content + + def delete_dataset( + self, + *, + dataset_uuid: str, + dataset_type: ModelTargetDatasetType, + ) -> None: + """Delete a Model Target dataset. + + Args: + dataset_uuid: The UUID of the dataset, as given by + :meth:`create_dataset`. + dataset_type: The kind of dataset to delete. + + Raises: + ~vws.exceptions.model_target_exceptions.ModelTargetAuthenticationError: + The request was not authenticated. + ~vws.exceptions.model_target_exceptions.UnknownModelTargetDatasetError: + No dataset of the given type matches the given UUID. + ~vws.exceptions.model_target_exceptions.ModelTargetOAuth2Error: + Vuforia did not give an access token. + """ + self.make_request( + method=HTTPMethod.DELETE, + data=b"", + request_path=dataset_path( + dataset_type=dataset_type, + dataset_uuid=dataset_uuid, + ), + ) diff --git a/src/vws/reports.py b/src/vws/reports.py index 66a9685ed..2be447b03 100644 --- a/src/vws/reports.py +++ b/src/vws/reports.py @@ -217,6 +217,138 @@ def from_response_dict(cls, response_dict: dict[str, Any]) -> Self: ) +@beartype +@unique +class ModelTargetDatasetStatuses(Enum): + """Constants representing Model Target dataset generation statuses. + + See the 'status' field of the dataset status response at + https://developer.vuforia.com/library/vuforia-engine/web-api/model-target-web-api/. + """ + + PROCESSING = "processing" + DONE = "done" + FAILED = "failed" + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetGenerationDetail: + """One detail of a Model Target dataset generation warning.""" + + code: str + message: str + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetGenerationError: + """The reason a Model Target dataset failed to generate.""" + + code: str + message: str + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetGenerationWarning: + """A warning about a generated Model Target dataset. + + A dataset with a warning is generated, and can be downloaded. + """ + + code: str + message: str + target: str + details: Sequence[ModelTargetGenerationDetail] + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetDatasetStatusReport: + """The status of a Model Target dataset. + + See + https://developer.vuforia.com/library/vuforia-engine/web-api/model-target-web-api/. + """ + + status: ModelTargetDatasetStatuses + dataset_uuid: str + created_at: datetime.datetime + eta: datetime.datetime | None + """When Vuforia expects to finish generating the dataset. + + This is given only while the dataset is processing. + """ + + completed_at: datetime.datetime | None + """When Vuforia finished generating the dataset. + + This is given only once the dataset is no longer processing. + """ + + error: ModelTargetGenerationError | None + """Why the dataset failed to generate. + + This is given only for a failed dataset. + """ + + warning: ModelTargetGenerationWarning | None + """A warning about the generated dataset. + + This is given only for a generated dataset which has a warning. + """ + + @classmethod + def from_response_dict(cls, response_dict: dict[str, Any]) -> Self: + """Construct from a Model Target Web API response dict.""" + error: ModelTargetGenerationError | None = None + if "error" in response_dict: + error_dict = dict(response_dict["error"]) + error = ModelTargetGenerationError( + code=error_dict["code"], + message=error_dict["message"], + ) + + warning: ModelTargetGenerationWarning | None = None + if "warning" in response_dict: + warning_dict = dict(response_dict["warning"]) + warning = ModelTargetGenerationWarning( + code=warning_dict["code"], + message=warning_dict["message"], + target=warning_dict["target"], + details=[ + ModelTargetGenerationDetail( + code=detail["code"], + message=detail["message"], + ) + for detail in warning_dict["details"] + ], + ) + + eta: datetime.datetime | None = None + if "eta" in response_dict: + eta = datetime.datetime.fromisoformat(response_dict["eta"]) + + completed_at: datetime.datetime | None = None + if "completedAt" in response_dict: + completed_at = datetime.datetime.fromisoformat( + response_dict["completedAt"], + ) + + return cls( + status=ModelTargetDatasetStatuses(value=response_dict["status"]), + dataset_uuid=response_dict["uuid"], + created_at=datetime.datetime.fromisoformat( + response_dict["createdAt"], + ), + eta=eta, + completed_at=completed_at, + error=error, + warning=warning, + ) + + @beartype @dataclass(frozen=True, kw_only=True) class RecoCount: diff --git a/tests/conftest.py b/tests/conftest.py index d94099ff8..74d4137be 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,11 +15,24 @@ from vws import ( VWS, AsyncCloudRecoService, + AsyncModelTargetService, AsyncVuMarkService, AsyncVWS, CloudRecoService, + ModelTargetService, VuMarkService, ) +from vws.model_target_datasets import ( + CadDataFormat, + GuideViewPosition, + ModelTargetModel, + ModelTargetView, +) + +# The mock accepts one hard-coded pair of Model Target Web API OAuth2 +# credentials, which it does not expose. +_MODEL_TARGET_CLIENT_ID = "client-id" +_MODEL_TARGET_CLIENT_SECRET = "client-secret" # noqa: S105 @pytest.fixture(name="_mock_database") @@ -124,6 +137,62 @@ async def async_vumark_service_client( yield client +@pytest.fixture(name="_mock_model_targets") +def fixture_mock_model_targets() -> Generator[None]: + """Yield a mock which serves the Model Target Web API. + + The Model Target Web API is not tied to a VWS database, so no + database is added. + """ + # We use a low processing time so that tests run quickly. + with MockVWS(processing_time_seconds=0.2): + yield + + +@pytest.fixture +def model_target_client( + *, + _mock_model_targets: None, +) -> ModelTargetService: + """A ``ModelTargetService`` client which connects to a mock.""" + return ModelTargetService( + client_id=_MODEL_TARGET_CLIENT_ID, + client_secret=_MODEL_TARGET_CLIENT_SECRET, + ) + + +@pytest_asyncio.fixture +async def async_model_target_client( + *, + _mock_model_targets: None, +) -> AsyncGenerator[AsyncModelTargetService]: + """An async ``ModelTargetService`` client which connects to a mock.""" + async with AsyncModelTargetService( + client_id=_MODEL_TARGET_CLIENT_ID, + client_secret=_MODEL_TARGET_CLIENT_SECRET, + ) as client: + yield client + + +@pytest.fixture(name="model_target_model") +def fixture_model_target_model() -> ModelTargetModel: + """A model which Vuforia accepts for dataset creation.""" + return ModelTargetModel( + name="model", + cad_data_url="https://example.com/model.zip", + cad_data_format=CadDataFormat.ZIP, + views=[ + ModelTargetView( + name="front", + guide_view_position=GuideViewPosition( + rotation=[0.0, 0.0, 0.0, 1.0], + translation=[0.0, 0.0, 1.0], + ), + ), + ], + ) + + @pytest.fixture(name="current_month") def fixture_current_month() -> datetime.date: """The current month, as the first day of that month.""" diff --git a/tests/test_async_model_targets.py b/tests/test_async_model_targets.py new file mode 100644 index 000000000..4bcd6a2f3 --- /dev/null +++ b/tests/test_async_model_targets.py @@ -0,0 +1,409 @@ +"""Tests for the async Model Target Web API client.""" + +import io +import json +import uuid +import zipfile +from http import HTTPStatus + +import pytest +from mock_vws import ( + MockVWS, + ModelTargetGenerationFailure, + ModelTargetGenerationWarning, +) + +from vws import AsyncModelTargetService +from vws.exceptions.model_target_exceptions import ( + ModelTargetDatasetNotDoneError, + ModelTargetDatasetTimeoutError, + ModelTargetOAuth2Error, + ModelTargetValidationError, + UnknownModelTargetDatasetError, +) +from vws.model_target_datasets import ( + CadDataFormat, + ModelTargetDatasetType, + ModelTargetModel, + RealisticAppearance, +) +from vws.reports import ModelTargetDatasetStatuses + +# The mock accepts one hard-coded pair of Model Target Web API OAuth2 +# credentials, which it does not expose. +_CLIENT_ID = "client-id" +_CLIENT_SECRET = "client-secret" # noqa: S105 + +_DATASET_TYPES = [ + ModelTargetDatasetType.STANDARD, + ModelTargetDatasetType.ADVANCED, +] + + +class TestAccessToken: + """Tests for getting an access token.""" + + @staticmethod + @pytest.mark.asyncio + async def test_token_is_a_bearer_token( + *, + async_model_target_client: AsyncModelTargetService, + ) -> None: + """An access token is given for valid credentials.""" + assert await async_model_target_client.get_access_token() + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.usefixtures("_mock_model_targets") + async def test_invalid_credentials() -> None: + """An exception is raised when the credentials are not known.""" + async with AsyncModelTargetService( + client_id="not-a-client-id", + client_secret="not-a-client-secret", # noqa: S106 + ) as client: + with pytest.raises( + expected_exception=ModelTargetOAuth2Error, + ) as exc: + await client.get_access_token() + + assert exc.value.response.status_code == HTTPStatus.UNAUTHORIZED + assert exc.value.error == "invalid_client" + + +class TestDatasetLifecycle: + """Tests for the dataset lifecycle.""" + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize( + argnames="dataset_type", + argvalues=_DATASET_TYPES, + ) + async def test_create_wait_download_delete( + *, + async_model_target_client: AsyncModelTargetService, + model_target_model: ModelTargetModel, + dataset_type: ModelTargetDatasetType, + ) -> None: + """A dataset can be created, downloaded and then deleted.""" + dataset_uuid = await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=dataset_type, + ) + + report = await async_model_target_client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + assert report.status == ModelTargetDatasetStatuses.DONE + assert report.dataset_uuid == dataset_uuid + assert report.completed_at is not None + assert report.eta is None + assert report.error is None + assert report.warning is None + + dataset = await async_model_target_client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + with zipfile.ZipFile( + file=io.BytesIO(initial_bytes=dataset) + ) as archive: + dataset_json = json.loads(s=archive.read(name="dataset.json")) + + assert dataset_json["uuid"] == dataset_uuid + assert dataset_json["type"] == dataset_type.value + + await async_model_target_client.delete_dataset( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + await async_model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + @staticmethod + @pytest.mark.asyncio + async def test_status_while_processing( + *, + async_model_target_client: AsyncModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A processing dataset has an estimated completion time.""" + dataset_uuid = await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = await async_model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.PROCESSING + assert report.eta is not None + assert report.completed_at is None + + @staticmethod + @pytest.mark.asyncio + async def test_download_while_processing( + *, + async_model_target_client: AsyncModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset cannot be downloaded before it is generated.""" + dataset_uuid = await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + with pytest.raises( + expected_exception=ModelTargetDatasetNotDoneError, + ) as exc: + await async_model_target_client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert ( + exc.value.response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + ) + assert exc.value.code == "UNSUPPORTED_STATE" + + @staticmethod + @pytest.mark.asyncio + async def test_dataset_types_are_separate( + *, + async_model_target_client: AsyncModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset is not visible to requests for the other type.""" + dataset_uuid = await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.ADVANCED, + ) + + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + await async_model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @staticmethod + @pytest.mark.asyncio + async def test_advanced_dataset_takes_multiple_models( + *, + async_model_target_client: AsyncModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """An advanced dataset can be generated from multiple models.""" + other_model = ModelTargetModel( + name="other-model", + cad_data_blob="ZmFrZS1jYWQtZGF0YQ==", + cad_data_format=CadDataFormat.GLB, + realistic_appearance=RealisticAppearance.TRUE, + ) + + assert await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model, other_model], + dataset_type=ModelTargetDatasetType.ADVANCED, + ) + + +class TestUnknownDataset: + """Tests for requests for datasets which do not exist.""" + + @staticmethod + @pytest.mark.asyncio + async def test_get_status( + *, + async_model_target_client: AsyncModelTargetService, + ) -> None: + """An exception is raised for an unknown dataset.""" + dataset_uuid = uuid.uuid4().hex + with pytest.raises( + expected_exception=UnknownModelTargetDatasetError, + ) as exc: + await async_model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert exc.value.response.status_code == HTTPStatus.NOT_FOUND + assert dataset_uuid in exc.value.message + + @staticmethod + @pytest.mark.asyncio + async def test_download( + *, + async_model_target_client: AsyncModelTargetService, + ) -> None: + """An exception is raised for an unknown dataset.""" + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + await async_model_target_client.download_dataset( + dataset_uuid=uuid.uuid4().hex, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @staticmethod + @pytest.mark.asyncio + async def test_delete( + *, + async_model_target_client: AsyncModelTargetService, + ) -> None: + """An exception is raised for an unknown dataset.""" + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + await async_model_target_client.delete_dataset( + dataset_uuid=uuid.uuid4().hex, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + +class TestValidation: + """Tests for requests which Vuforia rejects.""" + + @staticmethod + @pytest.mark.asyncio + async def test_no_cad_data( + *, + async_model_target_client: AsyncModelTargetService, + ) -> None: + """A model needs exactly one CAD data source.""" + with pytest.raises( + expected_exception=ModelTargetValidationError, + ) as exc: + await async_model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[ModelTargetModel(name="model")], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert exc.value.response.status_code == HTTPStatus.BAD_REQUEST + (detail,) = exc.value.details + assert detail.code == "VALIDATION_ERROR" + + +class TestGenerationResult: + """Tests for datasets which Vuforia does not generate cleanly.""" + + @staticmethod + @pytest.mark.asyncio + async def test_generation_failure( + *, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset which fails to generate reports the failure.""" + message = "Model Target dataset generation failed" + failure = ModelTargetGenerationFailure(message=message) + with MockVWS( + processing_time_seconds=0.2, + model_target_generation_failure=failure, + ): + async with AsyncModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) as client: + dataset_uuid = await client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = await client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.FAILED + assert report.error is not None + assert report.error.message == message + + @staticmethod + @pytest.mark.asyncio + async def test_generation_warning( + *, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset which generates with a warning reports the + warning. + """ + warning = ModelTargetGenerationWarning() + with MockVWS( + processing_time_seconds=0.2, + model_target_generation_warning=warning, + ): + async with AsyncModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) as client: + dataset_uuid = await client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = await client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.DONE + assert report.warning is not None + assert report.warning.target == dataset_uuid + (detail,) = report.warning.details + assert detail.code == "LOW_RECOGNITION_QUALITY" + + +class TestWaitForDatasetGenerated: + """Tests for waiting for a dataset to be generated.""" + + @staticmethod + @pytest.mark.asyncio + async def test_timeout(*, model_target_model: ModelTargetModel) -> None: + """An exception is raised when the wait times out.""" + with MockVWS(processing_time_seconds=60): + async with AsyncModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) as client: + dataset_uuid = await client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + with pytest.raises( + expected_exception=ModelTargetDatasetTimeoutError, + ): + await client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + seconds_between_requests=0.01, + timeout_seconds=0.05, + ) + + report = await client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.PROCESSING diff --git a/tests/test_model_targets.py b/tests/test_model_targets.py new file mode 100644 index 000000000..4ee13db6e --- /dev/null +++ b/tests/test_model_targets.py @@ -0,0 +1,802 @@ +"""Tests for the Model Target Web API client.""" + +import io +import json +import uuid +import zipfile +from http import HTTPStatus + +import pytest +from beartype import beartype +from freezegun import freeze_time +from mock_vws import ( + MockVWS, + ModelTargetGenerationFailure, + ModelTargetGenerationWarning, +) + +from vws import ModelTargetService +from vws.exceptions.model_target_exceptions import ( + ModelTargetAuthenticationError, + ModelTargetDatasetNotDoneError, + ModelTargetDatasetTimeoutError, + ModelTargetError, + ModelTargetOAuth2Error, + ModelTargetValidationError, + UnknownModelTargetDatasetError, +) +from vws.model_target_datasets import ( + CadDataFormat, + GuideViewPosition, + ModelTargetDatasetType, + ModelTargetModel, + ModelTargetView, + RealisticAppearance, +) +from vws.reports import ModelTargetDatasetStatuses +from vws.response import Response +from vws.transports import RequestsTransport, Transport + +# The mock accepts one hard-coded pair of Model Target Web API OAuth2 +# credentials, which it does not expose. +_CLIENT_ID = "client-id" +_CLIENT_SECRET = "client-secret" # noqa: S105 + +_DATASET_TYPES = [ + ModelTargetDatasetType.STANDARD, + ModelTargetDatasetType.ADVANCED, +] + + +@beartype +def _response(*, text: str) -> Response: + """Get a response with a given body. + + Args: + text: The body of the response. + + Returns: + A response with the given body. + """ + content = text.encode(encoding="utf-8") + return Response( + text=text, + url="https://vws.vuforia.com/modeltargets/datasets", + status_code=HTTPStatus.BAD_REQUEST, + headers={}, + request_body=None, + tell_position=len(content), + content=content, + ) + + +@beartype +class _CountingTransport: + """A transport which counts the requests made to each path.""" + + def __init__(self, *, transport: Transport) -> None: + """ + Args: + transport: The transport to make requests with. + """ + self._transport = transport + self.urls: list[str] = [] + + def close(self) -> None: + """Close the wrapped transport.""" + self._transport.close() + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + request_timeout: float | tuple[float, float], + ) -> Response: + """Make a request, recording the URL. + + Args: + method: The HTTP method. + url: The full URL. + headers: Request headers. + data: The request body. + request_timeout: The request timeout. + + Returns: + A Response populated from the HTTP response. + """ + self.urls.append(url) + return self._transport( + method=method, + url=url, + headers=headers, + data=data, + request_timeout=request_timeout, + ) + + +@beartype +class _BadTokenTransport: + """A transport which replaces each bearer token with an invalid + one. + """ + + def __init__(self, *, transport: Transport) -> None: + """ + Args: + transport: The transport to make requests with. + """ + self._transport = transport + + def close(self) -> None: + """Close the wrapped transport.""" + self._transport.close() + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + request_timeout: float | tuple[float, float], + ) -> Response: + """Make a request with an invalid bearer token. + + Args: + method: The HTTP method. + url: The full URL. + headers: Request headers. + data: The request body. + request_timeout: The request timeout. + + Returns: + A Response populated from the HTTP response. + """ + given_headers = dict(headers) + authorization = given_headers.get("Authorization", "") + if authorization.startswith("Bearer "): + given_headers["Authorization"] = "Bearer not-a-json-web-token" + + return self._transport( + method=method, + url=url, + headers=given_headers, + data=data, + request_timeout=request_timeout, + ) + + +class TestAccessToken: + """Tests for getting an access token.""" + + @staticmethod + @pytest.mark.usefixtures("_mock_model_targets") + def test_token_is_a_bearer_token() -> None: + """An access token is given for valid credentials.""" + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) + + assert client.get_access_token() + + @staticmethod + @pytest.mark.usefixtures("_mock_model_targets") + def test_token_is_reused( + *, + model_target_model: ModelTargetModel, + ) -> None: + """One access token is used for multiple requests.""" + transport = _CountingTransport(transport=RequestsTransport()) + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + transport=transport, + ) + + for _ in range(2): + client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + token_urls = [url for url in transport.urls if "oauth2" in url] + assert len(token_urls) == 1 + transport.close() + + @staticmethod + @pytest.mark.usefixtures("_mock_model_targets") + def test_expired_token_is_replaced( + *, + model_target_model: ModelTargetModel, + ) -> None: + """A new access token is requested once the old one expires.""" + transport = _CountingTransport(transport=RequestsTransport()) + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + transport=transport, + ) + + with freeze_time(time_to_freeze="2026-01-01") as frozen_time: + client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + # Mock tokens last an hour. + frozen_time.tick(delta=60 * 60 + 1) + client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + token_urls = [url for url in transport.urls if "oauth2" in url] + expected_token_request_count = 2 + assert len(token_urls) == expected_token_request_count + + @staticmethod + @pytest.mark.usefixtures("_mock_model_targets") + def test_invalid_credentials() -> None: + """An exception is raised when the credentials are not known.""" + client = ModelTargetService( + client_id="not-a-client-id", + client_secret="not-a-client-secret", # noqa: S106 + ) + + with pytest.raises( + expected_exception=ModelTargetOAuth2Error, + ) as exc: + client.get_access_token() + + assert exc.value.response.status_code == HTTPStatus.UNAUTHORIZED + assert exc.value.error == "invalid_client" + assert not exc.value.error_description + + @staticmethod + @pytest.mark.usefixtures("_mock_model_targets") + def test_invalid_bearer_token( + *, + model_target_model: ModelTargetModel, + ) -> None: + """An exception is raised when the bearer token is not + accepted. + """ + transport = _BadTokenTransport(transport=RequestsTransport()) + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + transport=transport, + ) + + with pytest.raises( + expected_exception=ModelTargetAuthenticationError, + ) as exc: + client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert exc.value.response.status_code == HTTPStatus.UNAUTHORIZED + assert exc.value.target == "jwt" + assert exc.value.message + transport.close() + + +class TestDatasetLifecycle: + """Tests for the dataset lifecycle.""" + + @staticmethod + @pytest.mark.parametrize( + argnames="dataset_type", + argvalues=_DATASET_TYPES, + ) + def test_create_wait_download_delete( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + dataset_type: ModelTargetDatasetType, + ) -> None: + """A dataset can be created, downloaded and then deleted.""" + dataset_uuid = model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=dataset_type, + ) + + report = model_target_client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + assert report.status == ModelTargetDatasetStatuses.DONE + assert report.dataset_uuid == dataset_uuid + assert report.completed_at is not None + assert report.completed_at >= report.created_at + assert report.eta is None + assert report.error is None + assert report.warning is None + + dataset = model_target_client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + with zipfile.ZipFile( + file=io.BytesIO(initial_bytes=dataset) + ) as archive: + dataset_json = json.loads(s=archive.read(name="dataset.json")) + + assert dataset_json["uuid"] == dataset_uuid + assert dataset_json["type"] == dataset_type.value + + model_target_client.delete_dataset( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=dataset_type, + ) + + @staticmethod + def test_status_while_processing( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A processing dataset has an estimated completion time.""" + dataset_uuid = model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.PROCESSING + assert report.eta is not None + assert report.eta >= report.created_at + assert report.completed_at is None + + @staticmethod + def test_download_while_processing( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset cannot be downloaded before it is generated.""" + dataset_uuid = model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + with pytest.raises( + expected_exception=ModelTargetDatasetNotDoneError, + ) as exc: + model_target_client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert ( + exc.value.response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + ) + assert exc.value.code == "UNSUPPORTED_STATE" + assert exc.value.target == dataset_uuid + + @staticmethod + def test_dataset_types_are_separate( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset is not visible to requests for the other type.""" + dataset_uuid = model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.ADVANCED, + ) + + @staticmethod + def test_advanced_dataset_takes_multiple_models( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """An advanced dataset can be generated from multiple models.""" + other_model = ModelTargetModel( + name="other-model", + cad_data_blob="ZmFrZS1jYWQtZGF0YQ==", + cad_data_format=CadDataFormat.GLB, + realistic_appearance=RealisticAppearance.TRUE, + ) + + dataset_uuid = model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model, other_model], + dataset_type=ModelTargetDatasetType.ADVANCED, + ) + + assert dataset_uuid + + @staticmethod + def test_state_based_model( + *, + model_target_client: ModelTargetService, + ) -> None: + """A State-Based Model Target dataset can be created.""" + configuration = json.dumps(obj={"states": {"open": {}, "closed": {}}}) + model = ModelTargetModel( + name="model", + cad_data_url="https://example.com/model.zip", + cad_data_format=CadDataFormat.ZIP, + state_based_configuration_json_string=configuration, + views=[ + ModelTargetView( + name="front", + guide_view_position=GuideViewPosition( + rotation=[0.0, 0.0, 0.0, 1.0], + translation=[0.0, 0.0, 1.0], + ), + states=["open"], + ), + ], + ) + + assert model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + +class TestUnknownDataset: + """Tests for requests for datasets which do not exist.""" + + @staticmethod + def test_get_status(*, model_target_client: ModelTargetService) -> None: + """An exception is raised for an unknown dataset.""" + dataset_uuid = uuid.uuid4().hex + with pytest.raises( + expected_exception=UnknownModelTargetDatasetError, + ) as exc: + model_target_client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert exc.value.response.status_code == HTTPStatus.NOT_FOUND + assert exc.value.code == "NOT_FOUND" + assert dataset_uuid in exc.value.message + + @staticmethod + def test_download(*, model_target_client: ModelTargetService) -> None: + """An exception is raised for an unknown dataset.""" + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + model_target_client.download_dataset( + dataset_uuid=uuid.uuid4().hex, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @staticmethod + def test_delete(*, model_target_client: ModelTargetService) -> None: + """An exception is raised for an unknown dataset.""" + with pytest.raises(expected_exception=UnknownModelTargetDatasetError): + model_target_client.delete_dataset( + dataset_uuid=uuid.uuid4().hex, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + +class TestValidation: + """Tests for requests which Vuforia rejects.""" + + @staticmethod + def test_no_cad_data( + *, + model_target_client: ModelTargetService, + ) -> None: + """A model needs exactly one CAD data source.""" + with pytest.raises( + expected_exception=ModelTargetValidationError, + ) as exc: + model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[ModelTargetModel(name="model")], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert exc.value.response.status_code == HTTPStatus.BAD_REQUEST + assert exc.value.code == "BAD_REQUEST" + (detail,) = exc.value.details + assert detail.code == "VALIDATION_ERROR" + assert "cadDataUrl" in detail.message + + @staticmethod + def test_two_cad_data_sources( + *, + model_target_client: ModelTargetService, + ) -> None: + """A model cannot give two CAD data sources.""" + model = ModelTargetModel( + name="model", + cad_data_url="https://example.com/model.zip", + cad_data_blob="ZmFrZS1jYWQtZGF0YQ==", + ) + + with pytest.raises(expected_exception=ModelTargetValidationError): + model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @staticmethod + def test_two_models_in_a_standard_dataset( + *, + model_target_client: ModelTargetService, + model_target_model: ModelTargetModel, + ) -> None: + """A standard dataset takes exactly one model.""" + with pytest.raises( + expected_exception=ModelTargetValidationError, + ) as exc: + model_target_client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model, model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + (detail,) = exc.value.details + assert detail.message == "exactly one model should be provided" + + +class TestGenerationResult: + """Tests for datasets which Vuforia does not generate cleanly.""" + + @staticmethod + def test_generation_failure( + *, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset which fails to generate reports the failure.""" + message = "Model Target dataset generation failed" + failure = ModelTargetGenerationFailure(message=message) + with MockVWS( + processing_time_seconds=0.2, + model_target_generation_failure=failure, + ): + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) + dataset_uuid = client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.FAILED + assert report.error is not None + assert report.error.message == message + assert report.warning is None + + with pytest.raises( + expected_exception=ModelTargetDatasetNotDoneError, + ): + client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @staticmethod + def test_generation_warning( + *, + model_target_model: ModelTargetModel, + ) -> None: + """A dataset which generates with a warning reports the + warning. + """ + warning = ModelTargetGenerationWarning() + with MockVWS( + processing_time_seconds=0.2, + model_target_generation_warning=warning, + ): + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) + dataset_uuid = client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.status == ModelTargetDatasetStatuses.DONE + assert report.error is None + assert report.warning is not None + assert report.warning.message == warning.message + assert report.warning.target == dataset_uuid + (detail,) = report.warning.details + assert detail.code == "LOW_RECOGNITION_QUALITY" + + assert client.download_dataset( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + +class TestWaitForDatasetGenerated: + """Tests for waiting for a dataset to be generated.""" + + @staticmethod + def test_timeout(*, model_target_model: ModelTargetModel) -> None: + """An exception is raised when the wait times out.""" + with MockVWS(processing_time_seconds=60): + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + ) + dataset_uuid = client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + with pytest.raises( + expected_exception=ModelTargetDatasetTimeoutError, + ): + client.wait_for_dataset_generated( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + seconds_between_requests=0.01, + timeout_seconds=0.05, + ) + + +class TestErrorEnvelope: + """Tests for reading responses which are not shaped like Model Target + Web API errors. + """ + + @staticmethod + @pytest.mark.parametrize( + argnames="text", + argvalues=[ + "", + "Not JSON", + "[]", + "{}", + '{"error": "not-an-object"}', + '{"transaction_id": "abc", "result_code": "Fail"}', + ], + ) + def test_unknown_error_shape(*, text: str) -> None: + """An error without a Model Target error object gives empty + values. + """ + error = ModelTargetError(response=_response(text=text)) + + assert not error.code + assert not error.message + assert not error.target + assert not error.details + + @staticmethod + def test_error_without_details() -> None: + """An error which gives no details has no details.""" + text = json.dumps(obj={"error": {"code": "ERROR", "message": "No"}}) + error = ModelTargetError(response=_response(text=text)) + + assert error.code == "ERROR" + assert error.message == "No" + assert not error.target + assert not error.details + + @staticmethod + @pytest.mark.parametrize( + argnames="text", + argvalues=["Not JSON", "[]", "{}"], + ) + def test_unknown_oauth2_error_shape(*, text: str) -> None: + """An OAuth2 error without an error code gives empty values.""" + error = ModelTargetOAuth2Error(response=_response(text=text)) + + assert not error.error + assert not error.error_description + + @staticmethod + def test_oauth2_error_description() -> None: + """An OAuth2 error description is given when Vuforia gives one.""" + description = "Missing or invalid authorization header" + text = json.dumps( + obj={ + "error": "invalid_request", + "error_description": description, + }, + ) + error = ModelTargetOAuth2Error(response=_response(text=text)) + + assert error.error == "invalid_request" + assert error.error_description == description + + +class TestBaseVWSURL: + """Tests for using a custom base URL.""" + + @staticmethod + def test_custom_base_url( + *, + model_target_model: ModelTargetModel, + ) -> None: + """The Model Target Web API can be served from a URL with a + path. + """ + base_vws_url = "https://example.com/vws" + with MockVWS( + base_vws_url=base_vws_url, + processing_time_seconds=0.2, + ): + client = ModelTargetService( + client_id=_CLIENT_ID, + client_secret=_CLIENT_SECRET, + base_vws_url=base_vws_url, + ) + dataset_uuid = client.create_dataset( + name="dataset", + target_sdk="11.0", + models=[model_target_model], + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + report = client.get_dataset_status( + dataset_uuid=dataset_uuid, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + assert report.dataset_uuid == dataset_uuid