diff --git a/conftest.py b/conftest.py index 1a2445a7..e4c92794 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 c2765245..f43793ce 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 f48730bf..baea7b49 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 ab7bfb0a..c8b5bdaf 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 00000000..5660e237 --- /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 7c7d50d6..da92a9be 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 74e7d385..09386f1d 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 a181cb81..b4d39cc0 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 00000000..c391342c --- /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 00000000..67a5e901 --- /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 00000000..8eaf0987 --- /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 00000000..95a9eaa1 --- /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 00000000..70b116d0 --- /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 66a9685e..2be447b0 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 d94099ff..74d4137b 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 00000000..4bcd6a2f --- /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 00000000..4ee13db6 --- /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