diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index c3af5d0b9..4918681f6 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -42,6 +42,9 @@ jobs: run: uv run --extra=dev prek run --all-files --hook-stage ${{ matrix.hook-stage }} --verbose env: + # Avoid intermittent uv distribution cache rename failures while + # prek installs hook environments on Windows. + UV_NO_CACHE: '1' UV_PYTHON: ${{ matrix.python-version }} - uses: pre-commit-ci/lite-action@v1.1.0 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 92b251db9..b830d99da 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -115,6 +115,7 @@ jobs: - tests/mock_vws/test_requests_mock_usage.py - tests/mock_vws/test_respx_mock_usage.py - tests/mock_vws/test_flask_app_usage.py + - tests/mock_vws/test_model_target_web_api.py - tests/mock_vws/test_vumark_generation_api.py - tests/mock_vws/test_target_validators.py - tests/mock_vws/test_docker.py diff --git a/README.rst b/README.rst index f49e6924f..1a96b097e 100644 --- a/README.rst +++ b/README.rst @@ -6,7 +6,7 @@ VWS Mock .. contents:: :local: -Mock for the Vuforia Web Services (VWS) API and the Vuforia Web Query API. +Mock for the Vuforia Web Services (VWS) API, the Vuforia Web Query API, and the Model Target Web API. Mocking calls made to Vuforia ------------------------------ diff --git a/admin/create_secrets_files.py b/admin/create_secrets_files.py index a3fdfaf5f..1cf5daf1d 100644 --- a/admin/create_secrets_files.py +++ b/admin/create_secrets_files.py @@ -12,7 +12,11 @@ import vws_web_tools from selenium.common.exceptions import TimeoutException from selenium.webdriver.remote.webdriver import WebDriver -from vws_web_tools import DatabaseDict, VuMarkDatabaseDict +from vws_web_tools import ( + DatabaseDict, + ModelTargetWebAPIDict, + VuMarkDatabaseDict, +) VUMARK_TEMPLATE_SVG_FILE_PATH = Path(__file__).with_name( name="vumark_template.svg", @@ -72,11 +76,13 @@ def _create_and_get_vumark_details( def _generate_secrets_file_content( + *, cloud_database_details: DatabaseDict, vumark_details: VuMarkDatabaseDict, inactive_database_details: DatabaseDict, inactive_vumark_details: VuMarkDatabaseDict, vumark_target_id: str, + model_target_web_api_details: ModelTargetWebAPIDict, ) -> str: """Generate the content of a secrets file.""" return textwrap.dedent( @@ -101,6 +107,10 @@ def _generate_secrets_file_content( INACTIVE_VUMARK_VUFORIA_TARGET_MANAGER_DATABASE_NAME={inactive_vumark_details["database_name"]} INACTIVE_VUMARK_VUFORIA_SERVER_ACCESS_KEY={inactive_vumark_details["server_access_key"]} INACTIVE_VUMARK_VUFORIA_SERVER_SECRET_KEY={inactive_vumark_details["server_secret_key"]} + + MODEL_TARGET_VUFORIA_CLIENT_ID={model_target_web_api_details["client_id"]} + MODEL_TARGET_VUFORIA_CLIENT_SECRET={model_target_web_api_details["client_secret"]} + MODEL_TARGET_VUFORIA_CAD_DATA_URL={model_target_web_api_details["cad_data_url"]} """, ) @@ -193,6 +203,21 @@ def _create_and_get_inactive_vumark_details( return vumark_database_details +def _get_model_target_web_api_details( + driver: WebDriver, + email_address: str, + password: str, +) -> ModelTargetWebAPIDict: + """Get credentials and input data for the Model Target Web API.""" + vws_web_tools.log_in( + driver=driver, + email_address=email_address, + password=password, + ) + vws_web_tools.wait_for_logged_in(driver=driver) + return vws_web_tools.get_model_target_web_api_details(driver=driver) + + def _create_vuforia_resource_names() -> tuple[str, str, str, str]: """Create names for Vuforia resources.""" time = datetime.datetime.now(tz=datetime.UTC).strftime( @@ -236,6 +261,14 @@ def main() -> None: ) inactive_vumark_driver.quit() + model_target_driver = vws_web_tools.create_chrome_driver() + model_target_web_api_details = _get_model_target_web_api_details( + driver=model_target_driver, + email_address=email_address, + password=password, + ) + model_target_driver.quit() + num_databases = 100 required_files = [ (new_secrets_dir / f"vuforia_secrets_{i}.env") @@ -291,6 +324,7 @@ def main() -> None: inactive_database_details=inactive_database_details, inactive_vumark_details=inactive_vumark_details, vumark_target_id=vumark_target_id, + model_target_web_api_details=model_target_web_api_details, ) file.write_text(data=file_contents) sys.stdout.write(f"Created database {file.name}\n") diff --git a/docs/source/differences-to-vws.rst b/docs/source/differences-to-vws.rst index 1f4876ea9..c2ae9a2c8 100644 --- a/docs/source/differences-to-vws.rst +++ b/docs/source/differences-to-vws.rst @@ -110,6 +110,13 @@ The mock returns a fixed minimal image in the requested format. The ``instance_id`` value is not encoded into the response image. Real Vuforia encodes the instance ID into the VuMark pattern. +Model Target datasets +--------------------- + +The Model Target Web API mock supports OAuth2 token requests, standard and advanced dataset creation, status polling, dataset downloads, and deletion. +The generated dataset download is a small valid zip file containing request metadata, not a real Vuforia Engine Model Target dataset. +Model Target API routes accept any non-empty bearer token. + Header cases ------------ diff --git a/newsfragments/2114.change b/newsfragments/2114.change new file mode 100644 index 000000000..e0ddb0890 --- /dev/null +++ b/newsfragments/2114.change @@ -0,0 +1 @@ +Add a mock implementation of the Model Target Web API, including OAuth2 token creation, standard and advanced dataset creation, status polling, dataset download, and deletion. diff --git a/pyproject.toml b/pyproject.toml index 1e2825c86..7f944c285 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -111,7 +111,7 @@ optional-dependencies.dev = [ "vulture==2.16", "vws-python==2026.2.25.1", "vws-test-fixtures==2023.3.5", - "vws-web-tools==2026.2.22.1", + "vws-web-tools==2026.5.21", "yamlfix==1.19.1", "zizmor==1.25.2", ] diff --git a/secrets.tar.gpg b/secrets.tar.gpg index 576c69750..41bcc48da 100644 Binary files a/secrets.tar.gpg and b/secrets.tar.gpg differ diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index d4ddc4a7e..b1f0fd74e 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -2,6 +2,7 @@ KiB MPixel MiB MissingSchema +OAuth Ubuntu VuMark admin diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index 9595a4a96..9704e6cad 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -27,7 +27,17 @@ TargetStatuses, ) from mock_vws._database_matchers import get_database_matching_server_keys -from mock_vws._mock_common import json_dump +from mock_vws._flask_server.target_manager import TARGET_MANAGER +from mock_vws._mock_common import RequestData, json_dump +from mock_vws._model_target_web_api import ( + create_model_target_dataset, + delete_model_target_dataset, + download_model_target_dataset, + get_model_target_dataset_status, +) +from mock_vws._model_target_web_api import ( + oauth2_token as model_target_oauth2_token, +) from mock_vws._services_validators import run_services_validators from mock_vws._services_validators.exceptions import ( FailError, @@ -44,7 +54,9 @@ ImageMatcher, StructuralSimilarityMatcher, ) +from mock_vws.model_target import ModelTargetDatasetType from mock_vws.target import ImageTarget +from mock_vws.target_manager import TargetManager from mock_vws.target_raters import ( HardcodedTargetTrackingRater, ) @@ -117,6 +129,32 @@ def get_all_vumark_databases() -> set[VuMarkDatabase]: } +@beartype +def _flask_request_data() -> RequestData: + """Return the current Flask request as shared request data.""" + return RequestData( + method=request.method, + path=request.path, + headers=dict(request.headers), + body=request.data, + ) + + +@beartype +def _model_target_manager() -> TargetManager: + """Return the target manager backing the Flask app.""" + return TARGET_MANAGER + + +@beartype +def _to_flask_response( + api_response: tuple[int, dict[str, str], str | bytes], +) -> Response: + """Convert a shared API response to a Flask response.""" + status_code, headers, body = api_response + return Response(response=body, status=status_code, headers=headers) + + @VWS_FLASK_APP.before_request @beartype def set_terminate_wsgi_input() -> None: @@ -154,6 +192,10 @@ def validate_request() -> None: """ if request.endpoint == "generate_vumark_instance": return + if request.path == "/oauth2/token" or request.path.startswith( + "/modeltargets/", + ): + return run_services_validators( request_headers=dict(request.headers), request_body=request.data, @@ -187,6 +229,157 @@ def handle_exceptions(exc: ValidatorError) -> Response: return response +@VWS_FLASK_APP.route(rule="/oauth2/token", methods=[HTTPMethod.POST]) +@beartype +def oauth2_token() -> Response: + """Obtain an OAuth2 token for the Model Target Web API.""" + return _to_flask_response( + api_response=model_target_oauth2_token( + request=_flask_request_data(), + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/datasets", + methods=[HTTPMethod.POST], +) +@beartype +def create_standard_model_target_dataset() -> Response: + """Create a standard Model Target dataset.""" + settings = VWSSettings.model_validate(obj={}) + return _to_flask_response( + api_response=create_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + processing_time_seconds=settings.processing_time_seconds, + dataset_type=ModelTargetDatasetType.STANDARD, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/advancedDatasets", + methods=[HTTPMethod.POST], +) +@beartype +def create_advanced_model_target_dataset() -> Response: + """Create an advanced Model Target dataset.""" + settings = VWSSettings.model_validate(obj={}) + return _to_flask_response( + api_response=create_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + processing_time_seconds=settings.processing_time_seconds, + dataset_type=ModelTargetDatasetType.ADVANCED, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/datasets//status", + methods=[HTTPMethod.GET], +) +@beartype +def get_standard_model_target_dataset_status( + dataset_uuid: str, +) -> Response: + """Return a standard Model Target dataset creation status.""" + return _to_flask_response( + api_response=get_model_target_dataset_status( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/advancedDatasets//status", + methods=[HTTPMethod.GET], +) +@beartype +def get_advanced_model_target_dataset_status( + dataset_uuid: str, +) -> Response: + """Return an advanced Model Target dataset creation status.""" + return _to_flask_response( + api_response=get_model_target_dataset_status( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/datasets//dataset", + methods=[HTTPMethod.GET], +) +@beartype +def download_standard_model_target_dataset( + dataset_uuid: str, +) -> Response: + """Download a standard Model Target dataset.""" + return _to_flask_response( + api_response=download_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/advancedDatasets//dataset", + methods=[HTTPMethod.GET], +) +@beartype +def download_advanced_model_target_dataset( + dataset_uuid: str, +) -> Response: + """Download an advanced Model Target dataset.""" + return _to_flask_response( + api_response=download_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/datasets/", + methods=[HTTPMethod.DELETE], +) +@beartype +def delete_standard_model_target_dataset(dataset_uuid: str) -> Response: + """Delete a standard Model Target dataset.""" + return _to_flask_response( + api_response=delete_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/modeltargets/advancedDatasets/", + methods=[HTTPMethod.DELETE], +) +@beartype +def delete_advanced_model_target_dataset(dataset_uuid: str) -> Response: + """Delete an advanced Model Target dataset.""" + return _to_flask_response( + api_response=delete_model_target_dataset( + request=_flask_request_data(), + target_manager=_model_target_manager(), + dataset_uuid=dataset_uuid, + ), + ) + + @VWS_FLASK_APP.route(rule="/targets", methods=[HTTPMethod.POST]) @beartype def add_target() -> Response: diff --git a/src/mock_vws/_model_target_web_api.py b/src/mock_vws/_model_target_web_api.py new file mode 100644 index 000000000..899d0bc6d --- /dev/null +++ b/src/mock_vws/_model_target_web_api.py @@ -0,0 +1,341 @@ +"""A fake implementation of the Model Target Web API.""" + +import base64 +import io +import json +import zipfile +from http import HTTPStatus +from typing import Any +from urllib.parse import parse_qs + +from beartype import beartype + +from mock_vws._mock_common import RequestData, json_dump +from mock_vws.model_target import ModelTargetDataset, ModelTargetDatasetType +from mock_vws.target_manager import TargetManager + +_ResponseType = tuple[int, dict[str, str], str | bytes] +_MAX_ADVANCED_MODEL_COUNT = 20 + + +@beartype +def _json_response( + *, + status_code: HTTPStatus, + body: dict[str, Any], +) -> _ResponseType: + """Return a JSON response.""" + body_json = json_dump(body=body) + return ( + status_code, + { + "Content-Length": str(object=len(body_json)), + "Content-Type": "application/json", + }, + body_json, + ) + + +@beartype +def _error_response( + *, + status_code: HTTPStatus, + code: str, + message: str, + target: str, +) -> _ResponseType: + """Return an error response shaped like the Model Target Web API.""" + return _json_response( + status_code=status_code, + body={ + "error": { + "code": code, + "message": message, + "target": target, + }, + }, + ) + + +@beartype +def _get_header(request: RequestData, name: str) -> str | None: + """Return a request header, case-insensitively.""" + lower_name = name.casefold() + for key, value in request.headers.items(): + if key.casefold() == lower_name: + return value + return None + + +@beartype +def _require_bearer_token(request: RequestData) -> _ResponseType | None: + """Return an error response if the request has no bearer token.""" + auth_header = _get_header(request=request, name="Authorization") + if auth_header is None or not auth_header.startswith("Bearer "): + return _error_response( + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="no Bearer token", + target="jwt", + ) + if not auth_header.removeprefix("Bearer ").strip(): + return _error_response( + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="invalid Bearer token", + target="jwt", + ) + return None + + +@beartype +def oauth2_token(request: RequestData) -> _ResponseType: + """Return a fake OAuth2 access token.""" + auth_header = _get_header(request=request, name="Authorization") + form = parse_qs(qs=request.body.decode(encoding="utf-8")) + grant_type = form.get("grant_type", [""])[0] + has_basic_auth = auth_header is not None and auth_header.startswith( + "Basic ", + ) + has_password_credentials = all( + form.get(field, [""])[0] for field in ("username", "password") + ) + if grant_type not in {"", "client_credentials", "password"} or ( + not has_basic_auth and not has_password_credentials + ): + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Invalid OAuth2 token request.", + target="grant_type", + ) + + token_source = request.body or (auth_header or "").encode() + access_token = base64.urlsafe_b64encode(s=token_source).decode( + encoding="ascii", + ) + access_token = access_token.rstrip("=") or "mock-vuforia-access-token" + return _json_response( + status_code=HTTPStatus.OK, + body={ + "access_token": access_token, + "token_type": "bearer", + "expires_in": 3600, + }, + ) + + +@beartype +def _load_request_json(request: RequestData) -> dict[str, Any] | _ResponseType: + """Load a Model Target dataset creation request body.""" + content_type = _get_header(request=request, name="Content-Type") or "" + if "application/json" not in content_type: + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Content-Type must be application/json.", + target="Content-Type", + ) + try: + request_json: dict[str, Any] = json.loads(s=request.body) + except json.JSONDecodeError: + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Request body must be valid JSON.", + target="body", + ) + return request_json + + +@beartype +def _validate_dataset_request( + *, + request_json: dict[str, Any], + dataset_type: ModelTargetDatasetType, +) -> _ResponseType | None: + """Validate the dataset request enough for useful mock feedback.""" + for field in ("name", "models", "targetSdk"): + if field not in request_json: + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message=f"Missing required field: {field}.", + target=field, + ) + + models_value = request_json["models"] + if not isinstance(models_value, list): + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="models must be a list.", + target="models", + ) + + models: list[Any] = [*models_value] + model_count = len(models) + + if dataset_type == ModelTargetDatasetType.STANDARD and model_count != 1: + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Standard Model Target datasets must have one model.", + target="models", + ) + + if ( + dataset_type == ModelTargetDatasetType.ADVANCED + and not 1 <= model_count <= _MAX_ADVANCED_MODEL_COUNT + ): + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Advanced Model Target datasets must have 1 to 20 models.", + target="models", + ) + + return None + + +@beartype +def create_model_target_dataset( + *, + request: RequestData, + target_manager: TargetManager, + processing_time_seconds: float, + dataset_type: ModelTargetDatasetType, +) -> _ResponseType: + """Create a standard or advanced Model Target dataset.""" + auth_error = _require_bearer_token(request=request) + if auth_error is not None: + return auth_error + + request_json_or_error = _load_request_json(request=request) + if not isinstance(request_json_or_error, dict): + return request_json_or_error + + validation_error = _validate_dataset_request( + request_json=request_json_or_error, + dataset_type=dataset_type, + ) + if validation_error is not None: + return validation_error + + dataset = ModelTargetDataset( + request_body=request_json_or_error, + dataset_type=dataset_type, + processing_time_seconds=processing_time_seconds, + ) + target_manager.add_model_target_dataset(model_target_dataset=dataset) + return _json_response( + status_code=HTTPStatus.CREATED, + body={"uuid": dataset.uuid_}, + ) + + +@beartype +def get_model_target_dataset_status( + *, + request: RequestData, + target_manager: TargetManager, + dataset_uuid: str, +) -> _ResponseType: + """Return the status of a Model Target dataset.""" + auth_error = _require_bearer_token(request=request) + if auth_error is not None: + return auth_error + try: + dataset = target_manager.model_target_datasets[dataset_uuid] + except KeyError: + return _error_response( + status_code=HTTPStatus.NOT_FOUND, + code="404", + message="The dataset was not found.", + target="uuid", + ) + return _json_response( + status_code=HTTPStatus.OK, + body=dataset.status_body(), + ) + + +@beartype +def _dataset_zip_bytes(dataset: ModelTargetDataset) -> bytes: + """Return a small valid zip file for a generated dataset.""" + zip_buffer = io.BytesIO() + with zipfile.ZipFile(file=zip_buffer, mode="w") as zip_file: + zip_file.writestr( + zinfo_or_arcname="dataset.json", + data=json.dumps( + obj={ + "uuid": dataset.uuid_, + "type": dataset.dataset_type.value, + "request": dataset.request_body, + }, + separators=(",", ":"), + ), + ) + return zip_buffer.getvalue() + + +@beartype +def download_model_target_dataset( + *, + request: RequestData, + target_manager: TargetManager, + dataset_uuid: str, +) -> _ResponseType: + """Download a generated Model Target dataset.""" + auth_error = _require_bearer_token(request=request) + if auth_error is not None: + return auth_error + try: + dataset = target_manager.model_target_datasets[dataset_uuid] + except KeyError: + return _error_response( + status_code=HTTPStatus.NOT_FOUND, + code="404", + message="The dataset was not found.", + target="uuid", + ) + if dataset.status != "done": + return _error_response( + status_code=HTTPStatus.UNPROCESSABLE_ENTITY, + code="UNPROCESSABLE_ENTITY", + message="The dataset is still processing.", + target="uuid", + ) + + body = _dataset_zip_bytes(dataset=dataset) + return ( + HTTPStatus.OK, + { + "Content-Length": str(object=len(body)), + "Content-Type": "application/zip", + }, + body, + ) + + +@beartype +def delete_model_target_dataset( + *, + request: RequestData, + target_manager: TargetManager, + dataset_uuid: str, +) -> _ResponseType: + """Delete a Model Target dataset.""" + auth_error = _require_bearer_token(request=request) + if auth_error is not None: + return auth_error + try: + target_manager.remove_model_target_dataset(dataset_uuid=dataset_uuid) + except KeyError: + return _error_response( + status_code=HTTPStatus.NOT_FOUND, + code="404", + message="The dataset was not found.", + target="uuid", + ) + return HTTPStatus.OK, {"Content-Length": "0"}, "" diff --git a/src/mock_vws/_requests_mock_server/mock_web_services_api.py b/src/mock_vws/_requests_mock_server/mock_web_services_api.py index 50fc7aa95..542dbc9cf 100644 --- a/src/mock_vws/_requests_mock_server/mock_web_services_api.py +++ b/src/mock_vws/_requests_mock_server/mock_web_services_api.py @@ -26,6 +26,13 @@ ) from mock_vws._database_matchers import get_database_matching_server_keys from mock_vws._mock_common import RequestData, Route, json_dump +from mock_vws._model_target_web_api import ( + create_model_target_dataset, + delete_model_target_dataset, + download_model_target_dataset, + get_model_target_dataset_status, + oauth2_token, +) from mock_vws._services_validators import run_services_validators from mock_vws._services_validators.exceptions import ( FailError, @@ -38,6 +45,7 @@ ) from mock_vws.database import VuMarkDatabase from mock_vws.image_matchers import ImageMatcher +from mock_vws.model_target import ModelTargetDatasetType from mock_vws.target import ImageTarget from mock_vws.target_manager import TargetManager from mock_vws.target_raters import TargetTrackingRater @@ -46,6 +54,7 @@ from mock_vws.database import CloudDatabase _TARGET_ID_PATTERN = "[A-Za-z0-9]+" +_MODEL_TARGET_DATASET_UUID_PATTERN = "[A-Za-z0-9-]+" _ROUTES: set[Route] = set() @@ -138,6 +147,159 @@ def __init__( self._duplicate_match_checker = duplicate_match_checker self._target_tracking_rater = target_tracking_rater + @route(path_pattern="/oauth2/token", http_methods={HTTPMethod.POST}) + def oauth2_token( # pylint: disable=no-self-use + self, + request: RequestData, + ) -> _ResponseType: + """Obtain an OAuth2 token for the Model Target Web API.""" + return oauth2_token(request=request) + + @route( + path_pattern="/modeltargets/datasets", + http_methods={HTTPMethod.POST}, + ) + def create_standard_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Create a standard Model Target dataset.""" + return create_model_target_dataset( + request=request, + target_manager=self._target_manager, + processing_time_seconds=self._processing_time_seconds, + dataset_type=ModelTargetDatasetType.STANDARD, + ) + + @route( + path_pattern="/modeltargets/advancedDatasets", + http_methods={HTTPMethod.POST}, + ) + def create_advanced_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Create an advanced Model Target dataset.""" + return create_model_target_dataset( + request=request, + target_manager=self._target_manager, + processing_time_seconds=self._processing_time_seconds, + dataset_type=ModelTargetDatasetType.ADVANCED, + ) + + @route( + path_pattern=( + "/modeltargets/datasets/" + f"{_MODEL_TARGET_DATASET_UUID_PATTERN}/status" + ), + http_methods={HTTPMethod.GET}, + ) + def get_standard_model_target_dataset_status( + self, + request: RequestData, + ) -> _ResponseType: + """Return a standard Model Target dataset creation status.""" + dataset_uuid = request.path.split(sep="/")[-2] + return get_model_target_dataset_status( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + + @route( + path_pattern=( + "/modeltargets/advancedDatasets/" + f"{_MODEL_TARGET_DATASET_UUID_PATTERN}/status" + ), + http_methods={HTTPMethod.GET}, + ) + def get_advanced_model_target_dataset_status( + self, + request: RequestData, + ) -> _ResponseType: + """Return an advanced Model Target dataset creation status.""" + dataset_uuid = request.path.split(sep="/")[-2] + return get_model_target_dataset_status( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + + @route( + path_pattern=( + "/modeltargets/datasets/" + f"{_MODEL_TARGET_DATASET_UUID_PATTERN}/dataset" + ), + http_methods={HTTPMethod.GET}, + ) + def download_standard_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Download a standard Model Target dataset.""" + dataset_uuid = request.path.split(sep="/")[-2] + return download_model_target_dataset( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + + @route( + path_pattern=( + "/modeltargets/advancedDatasets/" + f"{_MODEL_TARGET_DATASET_UUID_PATTERN}/dataset" + ), + http_methods={HTTPMethod.GET}, + ) + def download_advanced_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Download an advanced Model Target dataset.""" + dataset_uuid = request.path.split(sep="/")[-2] + return download_model_target_dataset( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + + @route( + path_pattern=( + f"/modeltargets/datasets/{_MODEL_TARGET_DATASET_UUID_PATTERN}" + ), + http_methods={HTTPMethod.DELETE}, + ) + def delete_standard_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Delete a standard Model Target dataset.""" + dataset_uuid = request.path.split(sep="/")[-1] + return delete_model_target_dataset( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + + @route( + path_pattern=( + "/modeltargets/advancedDatasets/" + f"{_MODEL_TARGET_DATASET_UUID_PATTERN}" + ), + http_methods={HTTPMethod.DELETE}, + ) + def delete_advanced_model_target_dataset( + self, + request: RequestData, + ) -> _ResponseType: + """Delete an advanced Model Target dataset.""" + dataset_uuid = request.path.split(sep="/")[-1] + return delete_model_target_dataset( + request=request, + target_manager=self._target_manager, + dataset_uuid=dataset_uuid, + ) + @route( path_pattern="/targets", http_methods={HTTPMethod.POST}, diff --git a/src/mock_vws/model_target.py b/src/mock_vws/model_target.py new file mode 100644 index 000000000..10ecb83bf --- /dev/null +++ b/src/mock_vws/model_target.py @@ -0,0 +1,79 @@ +"""Model Target dataset objects.""" + +import datetime +import uuid +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any +from zoneinfo import ZoneInfo + +from beartype import beartype + + +@beartype +class ModelTargetDatasetType(StrEnum): + """The kind of Model Target dataset.""" + + STANDARD = "standard" + ADVANCED = "advanced" + + +@beartype +def _now() -> datetime.datetime: + """Return the current time in UTC.""" + return datetime.datetime.now(tz=ZoneInfo(key="UTC")) + + +@beartype +def _format_datetime(value: datetime.datetime) -> str: + """Format a timestamp like the Model Target Web API.""" + return value.isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +@beartype +@dataclass(frozen=True, kw_only=True) +class ModelTargetDataset: + """A Model Target dataset generation request. + + Args: + request_body: The JSON request body used to start dataset creation. + dataset_type: Whether this is a standard or advanced dataset. + processing_time_seconds: The number of seconds before the generated + dataset becomes available. + uuid_: The dataset UUID. + created_at: When the dataset creation was requested. + """ + + request_body: dict[str, Any] = field(hash=False) + dataset_type: ModelTargetDatasetType + processing_time_seconds: float = field(hash=False) + uuid_: str = field(default_factory=lambda: uuid.uuid4().hex) + created_at: datetime.datetime = field(default_factory=_now) + + @property + def completed_at(self) -> datetime.datetime: + """When the dataset completes processing.""" + return self.created_at + datetime.timedelta( + seconds=self.processing_time_seconds, + ) + + @property + def status(self) -> str: + """The current dataset generation status.""" + if _now() < self.completed_at: + return "processing" + return "done" + + def status_body(self) -> dict[str, Any]: + """Return a status response body for this dataset.""" + body: dict[str, Any] = { + "status": self.status, + "uuid": self.uuid_, + "createdAt": _format_datetime(value=self.created_at), + } + if self.status == "processing": + body["eta"] = _format_datetime(value=self.completed_at) + else: + body["completedAt"] = _format_datetime(value=self.completed_at) + + return body diff --git a/src/mock_vws/target_manager.py b/src/mock_vws/target_manager.py index 14850df8e..24b78dcbd 100644 --- a/src/mock_vws/target_manager.py +++ b/src/mock_vws/target_manager.py @@ -5,6 +5,7 @@ from beartype import beartype from mock_vws.database import CloudDatabase, VuMarkDatabase +from mock_vws.model_target import ModelTargetDataset if TYPE_CHECKING: from mock_vws._database_matchers import AnyDatabase @@ -22,6 +23,7 @@ def __init__(self) -> None: """Create a target manager with no databases.""" self._cloud_databases: set[CloudDatabase] = set() self._vumark_databases: set[VuMarkDatabase] = set() + self._model_target_datasets: dict[str, ModelTargetDataset] = {} @property def cloud_databases(self) -> set[CloudDatabase]: @@ -33,6 +35,11 @@ def vumark_databases(self) -> set[VuMarkDatabase]: """All VuMark databases.""" return set(self._vumark_databases) + @property + def model_target_datasets(self) -> dict[str, ModelTargetDataset]: + """All Model Target datasets, keyed by UUID.""" + return dict(self._model_target_datasets) + def remove_cloud_database(self, cloud_database: CloudDatabase) -> None: """Remove a cloud database. @@ -56,6 +63,19 @@ def remove_vumark_database(self, vumark_database: VuMarkDatabase) -> None: db for db in self._vumark_databases if db != vumark_database } + def add_model_target_dataset( + self, + model_target_dataset: ModelTargetDataset, + ) -> None: + """Add a Model Target dataset.""" + self._model_target_datasets[model_target_dataset.uuid_] = ( + model_target_dataset + ) + + def remove_model_target_dataset(self, dataset_uuid: str) -> None: + """Remove a Model Target dataset.""" + del self._model_target_datasets[dataset_uuid] + def add_cloud_database(self, cloud_database: CloudDatabase) -> None: """Add a cloud database. diff --git a/tests/mock_vws/fixtures/credentials.py b/tests/mock_vws/fixtures/credentials.py index ba357b30d..0b187daf0 100644 --- a/tests/mock_vws/fixtures/credentials.py +++ b/tests/mock_vws/fixtures/credentials.py @@ -68,6 +68,20 @@ class _VuMarkCloudDatabaseSettings(BaseSettings): ) +class _ModelTargetSettings(BaseSettings): + """Settings for the Model Target Web API.""" + + client_id: str + client_secret: str + cad_data_url: str + + model_config = SettingsConfigDict( + env_prefix="MODEL_TARGET_VUFORIA_", + env_file=Path("vuforia_secrets.env"), + extra="allow", + ) + + @dataclass(frozen=True, kw_only=True) class InactiveVuMarkCloudDatabase: """Credentials for an inactive VuMark database.""" @@ -88,6 +102,27 @@ class VuMarkCloudDatabase: processing_target_id: str = field(repr=False) +@dataclass(frozen=True, kw_only=True) +class ModelTargetCredentials: + """Credentials and input data for the Model Target Web API.""" + + client_id: str = field(repr=False) + client_secret: str = field(repr=False) + cad_data_url: str = field(repr=False) + + +def get_model_target_credentials() -> ModelTargetCredentials: + """Return Model Target Web API credentials from environment + variables. + """ + settings = _ModelTargetSettings.model_validate(obj={}) + return ModelTargetCredentials( + client_id=settings.client_id, + client_secret=settings.client_secret, + cad_data_url=settings.cad_data_url, + ) + + @pytest.fixture def vuforia_database() -> CloudDatabase: """Return VWS credentials from environment variables.""" diff --git a/tests/mock_vws/fixtures/vuforia_backends.py b/tests/mock_vws/fixtures/vuforia_backends.py index adb281569..4300120de 100644 --- a/tests/mock_vws/fixtures/vuforia_backends.py +++ b/tests/mock_vws/fixtures/vuforia_backends.py @@ -261,6 +261,49 @@ def _enable_use_docker_in_memory( yield +@beartype +def _enable_use_real_model_target_vuforia( + *, + monkeypatch: pytest.MonkeyPatch, +) -> Generator[None]: + """Test against the real Model Target Web API.""" + assert monkeypatch + yield + + +@beartype +def _enable_use_mock_model_target_vuforia( + *, + monkeypatch: pytest.MonkeyPatch, +) -> Generator[None]: + """Test against the in-memory mock Model Target Web API.""" + assert monkeypatch + with MockVWS(): + yield + + +@beartype +def _enable_use_docker_in_memory_model_target_vuforia( + *, + monkeypatch: pytest.MonkeyPatch, +) -> Generator[None]: + """Test against the Flask-backed mock Model Target Web API.""" + assert monkeypatch + VWS_FLASK_APP.config["VWS_MOCK_TERMINATE_WSGI_INPUT"] = True + monkeypatch.setenv( + name="TARGET_MANAGER_BASE_URL", + value="http://example.com", + ) + + with responses.RequestsMock(assert_all_requests_are_fired=False) as mock: + add_flask_app_to_mock( + mock_obj=mock, + flask_app=VWS_FLASK_APP, + base_url="https://vws.vuforia.com", + ) + yield + + class VuforiaBackend(Enum): """Backends for tests.""" @@ -356,6 +399,40 @@ def fixture_verify_mock_vuforia( ) +@pytest.fixture( + name="verify_model_target_mock_vuforia", + params=list(VuforiaBackend), + ids=[backend.value for backend in list(VuforiaBackend)], +) +def fixture_verify_model_target_mock_vuforia( + *, + request: pytest.FixtureRequest, + monkeypatch: pytest.MonkeyPatch, +) -> Generator[VuforiaBackend]: + """Run Model Target Web API contract tests against real and mock + APIs. + """ + backend: VuforiaBackend = request.param + should_skip = request.config.getoption( + name=f"--skip-{backend.name.lower()}", + ) + if should_skip: + pytest.skip() + + enable_function = { + VuforiaBackend.REAL: _enable_use_real_model_target_vuforia, + VuforiaBackend.MOCK: _enable_use_mock_model_target_vuforia, + VuforiaBackend.DOCKER_IN_MEMORY: ( + _enable_use_docker_in_memory_model_target_vuforia + ), + }[backend] + + with contextlib.contextmanager(func=enable_function)( + monkeypatch=monkeypatch, + ): + yield backend + + @pytest.fixture( params=[item for item in VuforiaBackend if item != VuforiaBackend.REAL], ids=[ diff --git a/tests/mock_vws/test_docker.py b/tests/mock_vws/test_docker.py index 74d55b859..3bf78f074 100644 --- a/tests/mock_vws/test_docker.py +++ b/tests/mock_vws/test_docker.py @@ -52,7 +52,7 @@ def wait_for_health_check(container: Container) -> None: """Wait for a container to pass its health check. On failure, augment the error with the container's logs and the - Docker health check probe history so CI failures are diagnosable. + Docker health check probe history so CI failures are easier to diagnose. """ try: _poll_health_check(container=container) diff --git a/tests/mock_vws/test_flask_app_usage.py b/tests/mock_vws/test_flask_app_usage.py index 4db00b7e5..24388e60c 100644 --- a/tests/mock_vws/test_flask_app_usage.py +++ b/tests/mock_vws/test_flask_app_usage.py @@ -5,6 +5,7 @@ import json import time import uuid +import zipfile from collections.abc import Iterator from http import HTTPMethod, HTTPStatus @@ -30,6 +31,25 @@ ) _EXAMPLE_URL_FOR_TARGET_MANAGER = "http://" + uuid.uuid4().hex + ".com" +_MODEL_TARGET_DATASET_REQUEST = { + "name": "dataset-name", + "targetSdk": "10.18", + "models": [ + { + "name": "model-name", + "cadDataUrl": "https://example.com/model.glb", + "views": [ + { + "name": "view-name", + "guideViewPosition": { + "translation": [0, 0, 5], + "rotation": [0, 0, 0, 1], + }, + }, + ], + }, + ], +} @pytest.fixture(autouse=True) @@ -67,6 +87,8 @@ def _(*, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: TARGET_MANAGER.remove_cloud_database(cloud_database=cloud_database) for vumark_database in TARGET_MANAGER.vumark_databases: TARGET_MANAGER.remove_vumark_database(vumark_database=vumark_database) + for dataset_uuid in TARGET_MANAGER.model_target_datasets: + TARGET_MANAGER.remove_model_target_dataset(dataset_uuid=dataset_uuid) class TestProcessingTime: @@ -789,6 +811,57 @@ def test_processing_target_returns_forbidden() -> None: ) +class TestModelTargetWebAPI: + """Tests for the Model Target Web API through the Flask app.""" + + @staticmethod + def test_standard_dataset_workflow( + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """A Model Target dataset can be created and downloaded.""" + monkeypatch.setenv(name="PROCESSING_TIME_SECONDS", value="0") + token_response = requests.post( + url="https://vws.vuforia.com/oauth2/token", + auth=("client-id", "client-secret"), + data={"grant_type": "client_credentials"}, + timeout=30, + ) + token = token_response.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + + create_response = requests.post( + url="https://vws.vuforia.com/modeltargets/datasets", + headers=headers, + json=_MODEL_TARGET_DATASET_REQUEST, + timeout=30, + ) + dataset_uuid = create_response.json()["uuid"] + status_response = requests.get( + url=( + "https://vws.vuforia.com/modeltargets/datasets/" + f"{dataset_uuid}/status" + ), + headers=headers, + timeout=30, + ) + dataset_response = requests.get( + url=( + "https://vws.vuforia.com/modeltargets/datasets/" + f"{dataset_uuid}/dataset" + ), + headers=headers, + timeout=30, + ) + + assert token_response.status_code == HTTPStatus.OK + assert create_response.status_code == HTTPStatus.CREATED + assert status_response.json()["status"] == "done" + with zipfile.ZipFile( + file=io.BytesIO(initial_bytes=dataset_response.content), + ) as dataset_zip: + assert dataset_zip.namelist() == ["dataset.json"] + + class TestResponseDelay: """Tests for the response delay feature. diff --git a/tests/mock_vws/test_model_target_web_api.py b/tests/mock_vws/test_model_target_web_api.py new file mode 100644 index 000000000..fdd9059a5 --- /dev/null +++ b/tests/mock_vws/test_model_target_web_api.py @@ -0,0 +1,484 @@ +"""Verified fake tests for the Model Target Web API.""" + +import json +from http import HTTPMethod, HTTPStatus +from typing import Any +from uuid import uuid4 + +import pytest +import requests + +from mock_vws import MockVWS +from tests.mock_vws.fixtures.credentials import ( + ModelTargetCredentials, + get_model_target_credentials, +) +from tests.mock_vws.fixtures.vuforia_backends import VuforiaBackend + +_VWS_HOST = "https://vws.vuforia.com" +_DATASET_UUID = "0b12466eee5d49409a440927006ff5d8" + + +def _dataset_request(*, cad_data_url: str) -> dict[str, Any]: + """Return a standard Model Target dataset request.""" + return { + "name": f"dataset-{uuid4().hex}", + "targetSdk": "10.18", + "models": [ + { + "name": "model-name", + "cadDataUrl": cad_data_url, + "views": [ + { + "name": "view-name", + "guideViewPosition": { + "translation": [0, 0, 5], + "rotation": [0, 0, 0, 1], + }, + }, + ], + }, + ], + } + + +_UNAUTHENTICATED_DATASET_REQUEST = { + "name": "dataset-name", + "targetSdk": "10.18", + "models": [ + { + "name": "model-name", + "cadDataUrl": "https://example.com/model.glb", + "views": [ + { + "name": "view-name", + "guideViewPosition": { + "translation": [0, 0, 5], + "rotation": [0, 0, 0, 1], + }, + }, + ], + }, + ], +} + + +def _credentials_for_backend( + *, + backend: VuforiaBackend, +) -> ModelTargetCredentials: + """Return credentials for the chosen backend.""" + if backend == VuforiaBackend.REAL: + return get_model_target_credentials() + + return ModelTargetCredentials( + client_id="client-id", + client_secret="client-secret", + cad_data_url="https://example.com/model.glb", + ) + + +def _get_access_token(*, credentials: ModelTargetCredentials) -> str: + """Return an OAuth2 access token.""" + response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + auth=(credentials.client_id, credentials.client_secret), + data={"grant_type": "client_credentials"}, + timeout=30, + ) + + assert response.status_code == HTTPStatus.OK + response_json: dict[str, Any] = json.loads(s=response.text) + access_token = response_json["access_token"] + assert isinstance(access_token, str) + assert response_json["token_type"] == "bearer" + return access_token + + +def _assert_model_target_error( + *, + response: requests.Response, + status_code: HTTPStatus, + code: str, + message: str, + target: str, +) -> None: + """Assert a Model Target Web API error response.""" + assert response.status_code == status_code + assert response.json() == { + "error": { + "code": code, + "message": message, + "target": target, + }, + } + + +@pytest.mark.usefixtures("verify_model_target_mock_vuforia") +class TestAuthentication: + """Tests for Model Target Web API authentication.""" + + @staticmethod + @pytest.mark.parametrize( + argnames=("method", "path", "json_body"), + argvalues=[ + pytest.param( + HTTPMethod.POST, + "/modeltargets/datasets", + _UNAUTHENTICATED_DATASET_REQUEST, + id="create-standard-dataset", + ), + pytest.param( + HTTPMethod.POST, + "/modeltargets/advancedDatasets", + _UNAUTHENTICATED_DATASET_REQUEST, + id="create-advanced-dataset", + ), + pytest.param( + HTTPMethod.GET, + f"/modeltargets/datasets/{_DATASET_UUID}/status", + None, + id="standard-dataset-status", + ), + pytest.param( + HTTPMethod.GET, + f"/modeltargets/advancedDatasets/{_DATASET_UUID}/status", + None, + id="advanced-dataset-status", + ), + pytest.param( + HTTPMethod.GET, + f"/modeltargets/datasets/{_DATASET_UUID}/dataset", + None, + id="download-standard-dataset", + ), + pytest.param( + HTTPMethod.GET, + f"/modeltargets/advancedDatasets/{_DATASET_UUID}/dataset", + None, + id="download-advanced-dataset", + ), + pytest.param( + HTTPMethod.DELETE, + f"/modeltargets/datasets/{_DATASET_UUID}", + None, + id="delete-standard-dataset", + ), + pytest.param( + HTTPMethod.DELETE, + f"/modeltargets/advancedDatasets/{_DATASET_UUID}", + None, + id="delete-advanced-dataset", + ), + ], + ) + def test_missing_bearer_token( + *, + method: HTTPMethod, + path: str, + json_body: dict[str, object] | None, + ) -> None: + """Model Target routes require an OAuth2 bearer token.""" + response = requests.request( + method=method, + url=f"{_VWS_HOST}{path}", + json=json_body, + timeout=30, + ) + + assert response.status_code == HTTPStatus.UNAUTHORIZED + assert response.json() == { + "error": { + "code": "401", + "message": "no Bearer token", + "target": "jwt", + }, + } + + +class TestMockErrors: + """Tests for mock-only Model Target Web API error paths.""" + + @staticmethod + def test_invalid_oauth2_token_request() -> None: + """Invalid OAuth2 token requests are rejected.""" + with MockVWS(): + response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + data={"grant_type": "unsupported"}, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Invalid OAuth2 token request.", + target="grant_type", + ) + + @staticmethod + def test_blank_bearer_token() -> None: + """A blank bearer token is rejected.""" + with MockVWS(): + response = requests.get( + url=f"{_VWS_HOST}/modeltargets/datasets/{_DATASET_UUID}/status", + headers={"Authorization": "Bearer "}, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="invalid Bearer token", + target="jwt", + ) + + @staticmethod + @pytest.mark.parametrize( + argnames=("body", "headers", "message", "target"), + argvalues=[ + pytest.param( + "{}", + {}, + "Content-Type must be application/json.", + "Content-Type", + id="wrong-content-type", + ), + pytest.param( + "{", + {"Content-Type": "application/json"}, + "Request body must be valid JSON.", + "body", + id="invalid-json", + ), + ], + ) + def test_invalid_request_body( + *, + body: str, + headers: dict[str, str], + message: str, + target: str, + ) -> None: + """Invalid dataset request bodies are rejected.""" + with MockVWS(): + response = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={"Authorization": "Bearer token", **headers}, + data=body, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message=message, + target=target, + ) + + @staticmethod + @pytest.mark.parametrize( + argnames=("path", "body", "message", "target"), + argvalues=[ + pytest.param( + "/modeltargets/datasets", + {}, + "Missing required field: name.", + "name", + id="missing-name", + ), + pytest.param( + "/modeltargets/datasets", + { + "name": "dataset-name", + "targetSdk": "10.18", + "models": "model", + }, + "models must be a list.", + "models", + id="models-not-list", + ), + pytest.param( + "/modeltargets/datasets", + { + **_UNAUTHENTICATED_DATASET_REQUEST, + "models": [], + }, + "Standard Model Target datasets must have one model.", + "models", + id="standard-model-count", + ), + pytest.param( + "/modeltargets/advancedDatasets", + { + **_UNAUTHENTICATED_DATASET_REQUEST, + "models": [ + *_UNAUTHENTICATED_DATASET_REQUEST["models"], + ] + * 21, + }, + "Advanced Model Target datasets must have 1 to 20 models.", + "models", + id="advanced-model-count", + ), + ], + ) + def test_invalid_dataset_request( + *, + path: str, + body: dict[str, object], + message: str, + target: str, + ) -> None: + """Invalid dataset creation requests are rejected.""" + with MockVWS(): + response = requests.post( + url=f"{_VWS_HOST}{path}", + headers={"Authorization": "Bearer token"}, + json=body, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message=message, + target=target, + ) + + @staticmethod + @pytest.mark.parametrize( + argnames=("method", "path"), + argvalues=[ + pytest.param( + HTTPMethod.GET, + f"/modeltargets/datasets/{_DATASET_UUID}/status", + id="status", + ), + pytest.param( + HTTPMethod.GET, + f"/modeltargets/datasets/{_DATASET_UUID}/dataset", + id="download", + ), + pytest.param( + HTTPMethod.DELETE, + f"/modeltargets/datasets/{_DATASET_UUID}", + id="delete", + ), + ], + ) + def test_unknown_dataset( + *, + method: HTTPMethod, + path: str, + ) -> None: + """Unknown datasets are rejected.""" + with MockVWS(): + response = requests.request( + method=method, + url=f"{_VWS_HOST}{path}", + headers={"Authorization": "Bearer token"}, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.NOT_FOUND, + code="404", + message="The dataset was not found.", + target="uuid", + ) + + @staticmethod + def test_processing_dataset_cannot_be_downloaded() -> None: + """A dataset cannot be downloaded while it is still processing.""" + with MockVWS(processing_time_seconds=60): + create_response = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={"Authorization": "Bearer token"}, + json=_UNAUTHENTICATED_DATASET_REQUEST, + timeout=30, + ) + response = requests.get( + url=( + f"{_VWS_HOST}/modeltargets/datasets/" + f"{create_response.json()['uuid']}/dataset" + ), + headers={"Authorization": "Bearer token"}, + timeout=30, + ) + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.UNPROCESSABLE_ENTITY, + code="UNPROCESSABLE_ENTITY", + message="The dataset is still processing.", + target="uuid", + ) + + +class TestStandardDataset: + """Tests for standard Model Target datasets.""" + + @staticmethod + def test_create_status_and_delete( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """A standard Model Target dataset can be created and deleted.""" + credentials = _credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + access_token = _get_access_token(credentials=credentials) + headers = {"Authorization": f"Bearer {access_token}"} + dataset_uuid: str | None = None + + try: + create_response = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers=headers, + json=_dataset_request(cad_data_url=credentials.cad_data_url), + timeout=30, + ) + + assert create_response.status_code == HTTPStatus.CREATED + create_response_json: dict[str, Any] = json.loads( + s=create_response.text, + ) + dataset_uuid_value = create_response_json["uuid"] + assert isinstance(dataset_uuid_value, str) + dataset_uuid = dataset_uuid_value + + status_response = requests.get( + url=( + f"{_VWS_HOST}/modeltargets/datasets/{dataset_uuid}/status" + ), + headers=headers, + timeout=30, + ) + + assert status_response.status_code == HTTPStatus.OK + status_response_json: dict[str, Any] = json.loads( + s=status_response.text, + ) + assert status_response_json["status"] in { + "processing", + "done", + "failed", + } + assert isinstance(status_response_json["createdAt"], str) + finally: + if dataset_uuid is not None: # pragma: no branch + delete_response = requests.delete( + url=f"{_VWS_HOST}/modeltargets/datasets/{dataset_uuid}", + headers=headers, + timeout=30, + ) + assert delete_response.status_code in { + HTTPStatus.OK, + HTTPStatus.NO_CONTENT, + } diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index eaa82a494..34d6d8aa5 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -5,6 +5,7 @@ import io import json import socket +import zipfile from http import HTTPStatus from urllib.parse import urlparse @@ -26,6 +27,26 @@ processing_time_seconds, ) +_MODEL_TARGET_DATASET_REQUEST = { + "name": "dataset-name", + "targetSdk": "10.18", + "models": [ + { + "name": "model-name", + "cadDataUrl": "https://example.com/model.glb", + "views": [ + { + "name": "view-name", + "guideViewPosition": { + "translation": [0, 0, 5], + "rotation": [0, 0, 0, 1], + }, + }, + ], + }, + ], +} + @beartype def _not_exact_matcher( @@ -1052,3 +1073,95 @@ def test_httpx_real_http() -> None: pytest.raises(expected_exception=httpx.ConnectError), ): httpx.get(url=f"http://localhost:{port}", timeout=30) + + +class TestModelTargetWebAPI: + """Tests for the Model Target Web API.""" + + @staticmethod + def test_standard_dataset_workflow() -> None: + """A standard Model Target dataset can be created and + downloaded. + """ + with MockVWS(processing_time_seconds=0): + token_response = requests.post( + url="https://vws.vuforia.com/oauth2/token", + auth=("client-id", "client-secret"), + data={"grant_type": "client_credentials"}, + timeout=30, + ) + token = token_response.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + + create_response = requests.post( + url="https://vws.vuforia.com/modeltargets/datasets", + headers=headers, + json=_MODEL_TARGET_DATASET_REQUEST, + timeout=30, + ) + dataset_uuid = create_response.json()["uuid"] + + status_response = requests.get( + url=( + "https://vws.vuforia.com/modeltargets/datasets/" + f"{dataset_uuid}/status" + ), + headers=headers, + timeout=30, + ) + dataset_response = requests.get( + url=( + "https://vws.vuforia.com/modeltargets/datasets/" + f"{dataset_uuid}/dataset" + ), + headers=headers, + timeout=30, + ) + + assert token_response.status_code == HTTPStatus.OK + assert create_response.status_code == HTTPStatus.CREATED + assert status_response.json()["status"] == "done" + with zipfile.ZipFile( + file=io.BytesIO(initial_bytes=dataset_response.content), + ) as dataset_zip: + assert dataset_zip.namelist() == ["dataset.json"] + + @staticmethod + def test_advanced_dataset_workflow() -> None: + """An advanced Model Target dataset can be created.""" + with MockVWS(processing_time_seconds=0): + response = requests.post( + url="https://vws.vuforia.com/modeltargets/advancedDatasets", + headers={"Authorization": "Bearer token"}, + json=_MODEL_TARGET_DATASET_REQUEST, + timeout=30, + ) + dataset_uuid = response.json()["uuid"] + status_response = requests.get( + url=( + "https://vws.vuforia.com/modeltargets/" + f"advancedDatasets/{dataset_uuid}/status" + ), + headers={"Authorization": "Bearer token"}, + timeout=30, + ) + + assert response.status_code == HTTPStatus.CREATED + assert status_response.json()["uuid"] == dataset_uuid + + @staticmethod + def test_bearer_token_required() -> None: + """Model Target dataset routes require a bearer token.""" + with MockVWS(): + response = requests.post( + url="https://vws.vuforia.com/modeltargets/datasets", + json=_MODEL_TARGET_DATASET_REQUEST, + timeout=30, + ) + + assert response.status_code == HTTPStatus.UNAUTHORIZED + assert response.json()["error"] == { + "code": "401", + "message": "no Bearer token", + "target": "jwt", + } diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py index 5db88b2c5..467becf3f 100644 --- a/tests/mock_vws/test_respx_mock_usage.py +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -4,6 +4,7 @@ import io import uuid +from http import HTTPStatus import httpx import pytest @@ -18,6 +19,26 @@ from mock_vws.image_matchers import ExactMatcher from mock_vws.target import VuMarkTarget +_MODEL_TARGET_DATASET_REQUEST = { + "name": "dataset-name", + "targetSdk": "10.18", + "models": [ + { + "name": "model-name", + "cadDataUrl": "https://example.com/model.glb", + "views": [ + { + "name": "view-name", + "guideViewPosition": { + "translation": [0, 0, 5], + "rotation": [0, 0, 0, 1], + }, + }, + ], + }, + ], +} + class TestVWS: """Synchronous ``vws-python`` client usage through the mock via @@ -159,3 +180,30 @@ def test_generate_vumark_instance_returns_png_bytes() -> None: ) assert response_content.startswith(b"\x89PNG") + + +class TestModelTargetWebAPI: + """Model Target Web API usage through the mock via ``httpx``.""" + + @staticmethod + def test_standard_dataset_status() -> None: + """``httpx`` requests can use Model Target Web API routes.""" + with MockVWS(processing_time_seconds=0): + create_response = httpx.post( + url="https://vws.vuforia.com/modeltargets/datasets", + headers={"Authorization": "Bearer token"}, + json=_MODEL_TARGET_DATASET_REQUEST, + timeout=30, + ) + dataset_uuid = create_response.json()["uuid"] + status_response = httpx.get( + url=( + "https://vws.vuforia.com/modeltargets/datasets/" + f"{dataset_uuid}/status" + ), + headers={"Authorization": "Bearer token"}, + timeout=30, + ) + + assert create_response.status_code == HTTPStatus.CREATED + assert status_response.json()["status"] == "done" diff --git a/tests/mock_vws/test_target_validators.py b/tests/mock_vws/test_target_validators.py index 0dff2f865..0fc74601c 100644 --- a/tests/mock_vws/test_target_validators.py +++ b/tests/mock_vws/test_target_validators.py @@ -71,7 +71,7 @@ def test_validate_target_id_exists_uses_correct_path_segment( """ database = _database_with_target(target_id=target_id) - monkeypatch.setattr( # pylint: disable=bad-builtin + monkeypatch.setattr( target=target_validators, name="get_database_matching_server_keys", value=partial(_always_match_database, database=database), diff --git a/vuforia_secrets.env.example b/vuforia_secrets.env.example index 760e0407e..ae1990e24 100644 --- a/vuforia_secrets.env.example +++ b/vuforia_secrets.env.example @@ -24,3 +24,7 @@ INACTIVE_VUMARK_VUFORIA_TARGET_MANAGER_DATABASE_NAME=example_inactive_vumark_dat INACTIVE_VUMARK_VUFORIA_SERVER_ACCESS_KEY=example_inactive_vumark_server_access_key INACTIVE_VUMARK_VUFORIA_SERVER_SECRET_KEY=example_inactive_vumark_server_secret_key + +MODEL_TARGET_VUFORIA_CLIENT_ID=example_model_target_client_id +MODEL_TARGET_VUFORIA_CLIENT_SECRET=example_model_target_client_secret +MODEL_TARGET_VUFORIA_CAD_DATA_URL=https://example.com/model.glb