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

@@ -490,6 +490,8 @@ ignore_decorators = [
     "@*APP.route",
     "@*APP.before_request",
     "@*APP.errorhandler",
+    # requests-mock server
+    "@route",
 ]
 
 [tool.yamlfix]

spelling_private_dict.txt

@@ -100,6 +100,7 @@ resjsonarr
 rfc
 rgb
 str
+svg
 timestamp
 todo
 travis

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")
 
 
@@ -187,6 +198,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 +714,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 +726,97 @@ 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,
+            )
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        # Validate Accept header
+        try:
+            accept_header = validate_accept_header(
+                request_headers=request.headers,
+            )
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        # Extract and validate instance_id from request body
+        request_json: dict[str, Any] = json.loads(s=request.body or b"{}")
+        try:
+            instance_id = validate_instance_id(
+                instance_id=request_json.get("instance_id"),
+            )
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        # Look up the target and validate type and status
+        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)
+
+        try:
+            validate_target_type(target=target)
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        try:
+            validate_target_status_success(target=target)
+        except ValidatorError as exc:
+            return exc.status_code, exc.headers, exc.response_text
+
+        # 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-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