Add VuMark Generation API endpoint (#473)

.github/workflows/test.yml

@@ -111,6 +111,12 @@ jobs:
           - tests/mock_vws/test_update_target.py::TestUpdate
           - tests/mock_vws/test_update_target.py::TestWidth
           - tests/mock_vws/test_update_target.py::TestInactiveProject
+          - tests/mock_vws/test_vumark_generation.py::TestSuccessfulGeneration
+          - tests/mock_vws/test_vumark_generation.py::TestInvalidAcceptHeader
+          - tests/mock_vws/test_vumark_generation.py::TestInvalidInstanceId
+          - tests/mock_vws/test_vumark_generation.py::TestInvalidTargetType
+          - tests/mock_vws/test_vumark_generation.py::TestTargetStatusNotSuccess
+          - tests/mock_vws/test_vumark_generation.py::TestResponseHeaders
           - tests/mock_vws/test_requests_mock_usage.py
           - tests/mock_vws/test_flask_app_usage.py
           - tests/mock_vws/test_docker.py

pyproject.toml

@@ -478,6 +478,8 @@ ignore_names = [
     # Too difficult to test (see notes in the code)
     "DATE_RANGE_ERROR",
     "REQUEST_QUOTA_REACHED",
+    # requests-mock server route callback
+    "generate_vumark_instance",
     # pydantic-settings
     "model_config",
 ]

spelling_private_dict.txt

@@ -100,6 +100,7 @@ resjsonarr
 rfc
 rgb
 str
+svg
 timestamp
 todo
 travis
@@ -114,6 +115,7 @@ validators
 versioning
 vuforia
 vuforia's
+vumark
 vwq
 vws
 xa

src/mock_vws/_constants.py

@@ -38,6 +38,9 @@ class ResultCodes(Enum):
     PROJECT_INACTIVE = "ProjectInactive"
     INACTIVE_PROJECT = "InactiveProject"
     TOO_MANY_REQUESTS = "TooManyRequests"
+    INVALID_INSTANCE_ID = "InvalidInstanceId"
+    INVALID_ACCEPT_HEADER = "InvalidAcceptHeader"
+    INVALID_TARGET_TYPE = "InvalidTargetType"
 
 
 @beartype

src/mock_vws/_flask_server/target_manager.py

@@ -159,6 +159,10 @@ def create_database() -> Response:
         "state_name",
         random_database.state.name,
     )
+    default_target_type = request_json.get(
+        "default_target_type",
+        random_database.default_target_type,
+    )
 
     state = States[state_name]
 
@@ -168,6 +172,7 @@ def create_database() -> Response:
         client_access_key=client_access_key,
         client_secret_key=client_secret_key,
         database_name=database_name,
+        default_target_type=default_target_type,
         state=state,
     )
     try:
@@ -210,6 +215,7 @@ def create_target(database_name: str) -> Response:
         processing_time_seconds=request_json["processing_time_seconds"],
         application_metadata=request_json["application_metadata"],
         target_id=request_json["target_id"],
+        target_type=request_json.get("target_type", "cloud_target"),
         target_tracking_rater=target_tracking_rater,
     )
     database.targets.add(target)

src/mock_vws/_flask_server/vws.py

@@ -27,6 +27,17 @@
     TargetStatusProcessingError,
     ValidatorError,
 )
+from mock_vws._vumark_generators import (
+    generate_pdf,
+    generate_png,
+    generate_svg,
+)
+from mock_vws._vumark_validators import (
+    validate_accept_header,
+    validate_instance_id,
+    validate_target_status_success,
+    validate_target_type,
+)
 from mock_vws.database import VuforiaDatabase
 from mock_vws.image_matchers import (
     ExactMatcher,
@@ -181,6 +192,7 @@ def add_target() -> Response:
         processing_time_seconds=settings.processing_time_seconds,
         application_metadata=request_json.get("application_metadata"),
         target_tracking_rater=target_tracking_rater,
+        target_type=database.default_target_type,
     )
 
     databases_url = f"{settings.target_manager_base_url}/databases"
@@ -629,6 +641,74 @@ def update_target(target_id: str) -> Response:
     )
 
 
+@VWS_FLASK_APP.route(
+    rule="/targets/<string:target_id>/instances",
+    methods=[HTTPMethod.POST],
+)
+@beartype
+def generate_vumark_instance(target_id: str) -> Response:
+    """Generate a VuMark instance image.
+
+    Fake implementation of
+    https://developer.vuforia.com/library/vuforia-engine/web-api/vumark-generation-web-api/
+    """
+    databases = get_all_databases()
+    database = get_database_matching_server_keys(
+        request_headers=dict(request.headers),
+        request_body=request.data,
+        request_method=request.method,
+        request_path=request.path,
+        databases=databases,
+    )
+
+    # Validate Accept header
+    accept_header = validate_accept_header(
+        request_headers=dict(request.headers),
+    )
+
+    # Extract and validate instance_id from request body
+    request_json = json.loads(s=request.data)
+    instance_id = validate_instance_id(
+        instance_id=request_json.get("instance_id"),
+    )
+
+    # Verify target exists and validate type/status
+    (target,) = (
+        target for target in database.targets if target.target_id == target_id
+    )
+    validate_target_type(target=target)
+    validate_target_status_success(target=target)
+
+    # Generate the appropriate image format
+    if accept_header == "image/svg+xml":
+        content = generate_svg(instance_id=instance_id)
+        content_type = "image/svg+xml"
+    elif accept_header == "image/png":
+        content = generate_png(instance_id=instance_id)
+        content_type = "image/png"
+    else:  # PDF
+        content = generate_pdf(instance_id=instance_id)
+        content_type = "application/pdf"
+
+    date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True)
+    headers = {
+        "Connection": "keep-alive",
+        "Content-Type": content_type,
+        "server": "envoy",
+        "Date": date,
+        "x-envoy-upstream-service-time": "5",
+        "strict-transport-security": "max-age=31536000",
+        "x-aws-region": "us-east-2, us-west-2",
+        "x-content-type-options": "nosniff",
+    }
+
+    return Response(
+        status=HTTPStatus.OK,
+        response=content,
+        headers=headers,
+    )
+
+
 if __name__ == "__main__":  # pragma: no cover
     SETTINGS = VWSSettings.model_validate(obj={})
     VWS_FLASK_APP.run(host=SETTINGS.vws_host)

src/mock_vws/_requests_mock_server/mock_web_services_api.py

@@ -28,6 +28,17 @@
     TargetStatusProcessingError,
     ValidatorError,
 )
+from mock_vws._vumark_generators import (
+    generate_pdf,
+    generate_png,
+    generate_svg,
+)
+from mock_vws._vumark_validators import (
+    validate_accept_header,
+    validate_instance_id,
+    validate_target_status_success,
+    validate_target_type,
+)
 from mock_vws.image_matchers import ImageMatcher
 from mock_vws.target import Target
 from mock_vws.target_manager import TargetManager
@@ -38,7 +49,7 @@
 
 _ROUTES: set[Route] = set()
 
-_ResponseType = tuple[int, Mapping[str, str], str]
+_ResponseType = tuple[int, Mapping[str, str], str | bytes]
 _P = ParamSpec("_P")
 
 
@@ -104,6 +115,22 @@ def _body_bytes(request: PreparedRequest) -> bytes:
     return request.body
 
 
+@beartype
+def _generate_vumark_content(
+    *,
+    accept_header: str,
+    instance_id: str,
+) -> tuple[str, bytes]:
+    """Return generated VuMark content for the requested output format."""
+    generators: dict[str, Callable[[str], bytes]] = {
+        "image/svg+xml": generate_svg,
+        "image/png": generate_png,
+        "application/pdf": generate_pdf,
+    }
+    generator = generators[accept_header]
+    return accept_header, generator(instance_id)
+
+
 @beartype(conf=BeartypeConf(is_pep484_tower=True))
 class MockVuforiaWebServicesAPI:
     """A fake implementation of the Vuforia Web Services API.
@@ -187,6 +214,7 @@ def add_target(self, request: PreparedRequest) -> _ResponseType:
             processing_time_seconds=self._processing_time_seconds,
             application_metadata=application_metadata,
             target_tracking_rater=self._target_tracking_rater,
+            target_type=database.default_target_type,
         )
         database.targets.add(new_target)
 
@@ -702,7 +730,7 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType:
             "previous_month_recos": target.previous_month_recos,
         }
         body_json = json_dump(body=body)
-        headers = {
+        target_summary_headers = {
             "Connection": "keep-alive",
             "Content-Length": str(object=len(body_json)),
             "Content-Type": "application/json",
@@ -714,4 +742,72 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType:
             "x-content-type-options": "nosniff",
         }
 
-        return HTTPStatus.OK, headers, body_json
+        return HTTPStatus.OK, target_summary_headers, body_json
+
+    @route(
+        path_pattern=f"/targets/{_TARGET_ID_PATTERN}/instances",
+        http_methods={HTTPMethod.POST},
+    )
+    def generate_vumark_instance(
+        self,
+        request: PreparedRequest,
+    ) -> _ResponseType:
+        """Generate a VuMark instance image.
+
+        Fake implementation of
+        https://developer.vuforia.com/library/vuforia-engine/web-api/vumark-generation-web-api/
+        """
+        try:
+            run_services_validators(
+                request_headers=request.headers,
+                request_body=_body_bytes(request=request),
+                request_method=request.method or "",
+                request_path=request.path_url,
+                databases=self._target_manager.databases,
+            )
+            accept_header = validate_accept_header(
+                request_headers=request.headers,
+            )
+
+            request_json: dict[str, Any] = json.loads(s=request.body or b"{}")
+            instance_id = validate_instance_id(
+                instance_id=request_json.get("instance_id"),
+            )
+
+            database = get_database_matching_server_keys(
+                request_headers=request.headers,
+                request_body=_body_bytes(request=request),
+                request_method=request.method or "",
+                request_path=request.path_url,
+                databases=self._target_manager.databases,
+            )
+            split_path = request.path_url.split(sep="/")
+            target_id = split_path[-2]
+            target = database.get_target(target_id=target_id)
+            validate_target_type(target=target)
+            validate_target_status_success(target=target)
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        content_type, content = _generate_vumark_content(
+            accept_header=accept_header,
+            instance_id=instance_id,
+        )
+
+        date = email.utils.formatdate(
+            timeval=None,
+            localtime=False,
+            usegmt=True,
+        )
+        headers = {
+            "Connection": "keep-alive",
+            "Content-Length": str(object=len(content)),
+            "Content-Type": content_type,
+            "Date": date,
+            "server": "envoy",
+            "x-envoy-upstream-service-time": "5",
+            "strict-transport-security": "max-age=31536000",
+            "x-aws-region": "us-east-2, us-west-2",
+            "x-content-type-options": "nosniff",
+        }
+        return HTTPStatus.OK, headers, content