diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 386db8031..9a90a811e 100644 --- a/.github/workflows/test.yml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index 10f8cbc1f..7fd41f681 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index 365309073..08bbf6709 100644 --- a/spelling_private_dict.txt +++ b/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 diff --git a/src/mock_vws/_constants.py b/src/mock_vws/_constants.py index 1f832af1f..32e0cc582 100644 --- a/src/mock_vws/_constants.py +++ b/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 diff --git a/src/mock_vws/_flask_server/target_manager.py b/src/mock_vws/_flask_server/target_manager.py index 90f3341eb..282aad9c7 100644 --- a/src/mock_vws/_flask_server/target_manager.py +++ b/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) diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index 6463b928f..fe352e7f5 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/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//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) 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 ddee6ef03..2a7c831cf 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 @@ -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 diff --git a/src/mock_vws/_services_validators/exceptions.py b/src/mock_vws/_services_validators/exceptions.py index 4bbc5dab8..2a4f2396c 100644 --- a/src/mock_vws/_services_validators/exceptions.py +++ b/src/mock_vws/_services_validators/exceptions.py @@ -566,3 +566,121 @@ def __init__(self) -> None: "x-aws-region": "us-east-2, us-west-2", "x-content-type-options": "nosniff", } + + +@beartype +class InvalidInstanceIdError(ValidatorError): + """Exception raised when an invalid VuMark instance ID is given.""" + + def __init__(self) -> None: + """ + Attributes: + status_code: The status code to use in a response if this is + raised. + response_text: The response text to use in a response if this + is + raised. + """ + super().__init__() + self.status_code = HTTPStatus.UNPROCESSABLE_ENTITY + body = { + "transaction_id": uuid.uuid4().hex, + "result_code": ResultCodes.INVALID_INSTANCE_ID.value, + } + self.response_text = json_dump(body=body) + date = email.utils.formatdate( + timeval=None, + localtime=False, + usegmt=True, + ) + self.headers = { + "Connection": "keep-alive", + "Content-Type": "application/json", + "server": "envoy", + "Date": date, + "x-envoy-upstream-service-time": "5", + "Content-Length": str(object=len(self.response_text)), + "strict-transport-security": "max-age=31536000", + "x-aws-region": "us-east-2, us-west-2", + "x-content-type-options": "nosniff", + } + + +@beartype +class InvalidTargetTypeError(ValidatorError): + """Exception raised when a non-vumark target is used for vumark + generation. + """ + + def __init__(self) -> None: + """ + Attributes: + status_code: The status code to use in a response if this is + raised. + response_text: The response text to use in a response if this + is + raised. + """ + super().__init__() + self.status_code = HTTPStatus.UNPROCESSABLE_ENTITY + body = { + "transaction_id": uuid.uuid4().hex, + "result_code": ResultCodes.INVALID_TARGET_TYPE.value, + } + self.response_text = json_dump(body=body) + date = email.utils.formatdate( + timeval=None, + localtime=False, + usegmt=True, + ) + self.headers = { + "Connection": "keep-alive", + "Content-Type": "application/json", + "server": "envoy", + "Date": date, + "x-envoy-upstream-service-time": "5", + "Content-Length": str(object=len(self.response_text)), + "strict-transport-security": "max-age=31536000", + "x-aws-region": "us-east-2, us-west-2", + "x-content-type-options": "nosniff", + } + + +@beartype +class InvalidAcceptHeaderError(ValidatorError): + """Exception raised when an invalid Accept header is given for vumark + generation. + """ + + def __init__(self) -> None: + """ + Attributes: + status_code: The status code to use in a response if this is + raised. + response_text: The response text to use in a response if this + is + raised. + """ + super().__init__() + self.status_code = HTTPStatus.BAD_REQUEST + body = { + "transaction_id": uuid.uuid4().hex, + "result_code": ResultCodes.INVALID_ACCEPT_HEADER.value, + } + self.response_text = json_dump(body=body) + date = email.utils.formatdate( + timeval=None, + localtime=False, + usegmt=True, + ) + self.headers = { + "Connection": "keep-alive", + "Content-Type": "application/json", + "server": "envoy", + "Date": date, + "x-envoy-upstream-service-time": "5", + "Content-Length": str(object=len(self.response_text)), + "strict-transport-security": "max-age=31536000", + "x-aws-region": "us-east-2, us-west-2", + "x-content-type-options": "nosniff", + } diff --git a/src/mock_vws/_services_validators/key_validators.py b/src/mock_vws/_services_validators/key_validators.py index b07533fe0..b2f8ddeb5 100644 --- a/src/mock_vws/_services_validators/key_validators.py +++ b/src/mock_vws/_services_validators/key_validators.py @@ -121,6 +121,13 @@ def validate_keys( optional_keys=set(), ) + generate_vumark = _Route( + path_pattern=f"/targets/{target_id_pattern}/instances", + http_methods={HTTPMethod.POST}, + mandatory_keys={"instance_id"}, + optional_keys=set(), + ) + routes = ( add_target, delete_target, @@ -130,6 +137,7 @@ def validate_keys( get_duplicates, update_target, target_summary, + generate_vumark, ) (matching_route,) = ( diff --git a/src/mock_vws/_services_validators/target_validators.py b/src/mock_vws/_services_validators/target_validators.py index 4dbee04b1..a900e7d97 100644 --- a/src/mock_vws/_services_validators/target_validators.py +++ b/src/mock_vws/_services_validators/target_validators.py @@ -41,7 +41,15 @@ def validate_target_id_exists( if len(split_path) == request_path_no_target_id_length: return - target_id = split_path[-1] + # Handle only /targets/{id}/instances (VuMark generation) paths. + # For all other paths, the target ID is the final segment. + vumark_path_min_length = 4 + is_vumark_instances_path = ( + len(split_path) >= vumark_path_min_length + and split_path[-3] == "targets" + and split_path[-1] == "instances" + ) + target_id = split_path[-2] if is_vumark_instances_path else split_path[-1] database = get_database_matching_server_keys( request_headers=request_headers, request_body=request_body, diff --git a/src/mock_vws/_vumark_generators.py b/src/mock_vws/_vumark_generators.py new file mode 100644 index 000000000..407573932 --- /dev/null +++ b/src/mock_vws/_vumark_generators.py @@ -0,0 +1,129 @@ +"""Generate placeholder VuMark images for the mock API.""" + +import io +from xml.sax.saxutils import escape + +from beartype import beartype +from PIL import Image, ImageDraw + + +@beartype +def generate_svg(instance_id: str) -> bytes: + """Generate a placeholder SVG image for a VuMark instance. + + Args: + instance_id: The VuMark instance ID. + + Returns: + SVG image data as bytes. + """ + escaped_id = escape(data=instance_id) + svg_content = ( + '' + '' + '' + 'VuMark Mock' + '{escaped_id}' + "" + ) + return svg_content.encode() + + +@beartype +def generate_png(instance_id: str) -> bytes: + """Generate a placeholder PNG image for a VuMark instance. + + Args: + instance_id: The VuMark instance ID. + + Returns: + PNG image data as bytes. + """ + img = Image.new(mode="RGB", size=(200, 200), color=(255, 255, 255)) + draw = ImageDraw.Draw(im=img) + + draw.rectangle(xy=[0, 0, 199, 199], outline="black") + draw.text(xy=(100, 80), text="VuMark Mock", fill="black") + draw.text(xy=(100, 110), text=instance_id[:20], fill="black") + + buffer = io.BytesIO() + img.save(fp=buffer, format="PNG") + return buffer.getvalue() + + +@beartype +def generate_pdf(instance_id: str) -> bytes: + """Generate a placeholder PDF document for a VuMark instance. + + This creates a minimal valid PDF with the instance ID. + + Args: + instance_id: The VuMark instance ID. + + Returns: + PDF document data as bytes. + """ + # Escape parentheses in instance_id for PDF string literals. + safe_id = instance_id.replace("\\", "\\\\") + safe_id = safe_id.replace("(", "\\(") + safe_id = safe_id.replace(")", "\\)") + + # Build the stream content first so we can measure its length. + stream = ( + "BT\n" + "/F1 12 Tf\n" + "50 150 Td\n" + "(VuMark Mock) Tj\n" + "0 -20 Td\n" + f"({safe_id}) Tj\n" + "ET\n" + ) + stream_bytes = stream.encode() + stream_length = len(stream_bytes) + + # Build each object, tracking byte offsets for the xref table. + obj1 = "1 0 obj\n<< /Type /Catalog /Pages 2 0 R >>\nendobj\n" + obj2 = "2 0 obj\n<< /Type /Pages /Kids [3 0 R] /Count 1 >>\nendobj\n" + obj3 = ( + "3 0 obj\n" + "<< /Type /Page /Parent 2 0 R" + " /MediaBox [0 0 200 200]" + " /Contents 4 0 R" + " /Resources << /Font << /F1 5 0 R >> >> >>\n" + "endobj\n" + ) + obj4 = ( + f"4 0 obj\n<< /Length {stream_length} >>\n" + f"stream\n{stream}endstream\nendobj\n" + ) + obj5 = ( + "5 0 obj\n" + "<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>\n" + "endobj\n" + ) + + header = "%PDF-1.4\n" + offsets: list[int] = [] + body = header + for obj in (obj1, obj2, obj3, obj4, obj5): + offsets.append(len(body.encode())) + body += obj + + xref_offset = len(body.encode()) + xref = f"xref\n0 {len(offsets) + 1}\n0000000000 65535 f \n" + for offset in offsets: + xref += f"{offset:010d} 00000 n \n" + + trailer = ( + "trailer\n" + f"<< /Size {len(offsets) + 1} /Root 1 0 R >>\n" + "startxref\n" + f"{xref_offset}\n" + "%%EOF" + ) + + return (body + xref + trailer).encode() diff --git a/src/mock_vws/_vumark_validators/__init__.py b/src/mock_vws/_vumark_validators/__init__.py new file mode 100644 index 000000000..fab86ec76 --- /dev/null +++ b/src/mock_vws/_vumark_validators/__init__.py @@ -0,0 +1,90 @@ +"""Validators for VuMark generation requests.""" + +from collections.abc import Mapping + +from beartype import beartype + +from mock_vws._constants import TargetStatuses +from mock_vws._services_validators.exceptions import ( + InvalidAcceptHeaderError, + InvalidInstanceIdError, + InvalidTargetTypeError, + TargetStatusNotSuccessError, +) +from mock_vws.target import Target + +VALID_ACCEPT_HEADERS = frozenset( + {"image/svg+xml", "image/png", "application/pdf"} +) + + +@beartype +def validate_accept_header(request_headers: Mapping[str, str]) -> str: + """Validate the Accept header for VuMark generation. + + Args: + request_headers: The headers sent with the request. + + Returns: + The validated Accept header value. + + Raises: + InvalidAcceptHeaderError: The Accept header is missing or invalid. + """ + accept_header: str = request_headers.get("Accept") or "" + if accept_header not in VALID_ACCEPT_HEADERS: + raise InvalidAcceptHeaderError + return accept_header + + +@beartype +def validate_instance_id(instance_id: object) -> str: + """Validate the instance_id for VuMark generation. + + In the real Vuforia API, validation depends on the VuMark type: + - Numeric: 0-9 only + - Bytes: hex characters 0-9a-f + - String: printable ASCII characters + + For this mock, we accept any non-empty string. + + Args: + instance_id: The instance ID from the request body. + + Returns: + The validated instance ID. + + Raises: + InvalidInstanceIdError: The instance_id is missing or invalid. + """ + if not instance_id or not isinstance(instance_id, str): + raise InvalidInstanceIdError + return instance_id + + +@beartype +def validate_target_type(target: Target) -> None: + """Validate that the target is a VuMark target. + + Args: + target: The target to validate. + + Raises: + InvalidTargetTypeError: The target is not a VuMark target. + """ + if target.target_type != "vumark": + raise InvalidTargetTypeError + + +@beartype +def validate_target_status_success(target: Target) -> None: + """Validate that the target has a success status. + + Args: + target: The target to validate. + + Raises: + TargetStatusNotSuccessError: The target is not in success status. + """ + if target.status != TargetStatuses.SUCCESS.value: + raise TargetStatusNotSuccessError diff --git a/src/mock_vws/database.py b/src/mock_vws/database.py index 2e28a9f61..c7f8b4cc5 100644 --- a/src/mock_vws/database.py +++ b/src/mock_vws/database.py @@ -21,6 +21,7 @@ class DatabaseDict(TypedDict): server_secret_key: str client_access_key: str client_secret_key: str + default_target_type: str state_name: str targets: Iterable[TargetDict] @@ -62,6 +63,7 @@ class VuforiaDatabase: # In particular, we might want to inspect the ``database`` object's targets # as they change via API requests. targets: set[Target] = field(default_factory=set[Target], hash=False) + default_target_type: str = "cloud_target" state: States = States.WORKING request_quota: int = 100000 @@ -80,6 +82,7 @@ def to_dict(self) -> DatabaseDict: "server_secret_key": self.server_secret_key, "client_access_key": self.client_access_key, "client_secret_key": self.client_secret_key, + "default_target_type": self.default_target_type, "state_name": self.state.name, "targets": targets, } @@ -100,6 +103,7 @@ def from_dict(cls, database_dict: DatabaseDict) -> Self: server_secret_key=database_dict["server_secret_key"], client_access_key=database_dict["client_access_key"], client_secret_key=database_dict["client_secret_key"], + default_target_type=database_dict["default_target_type"], state=States[database_dict["state_name"]], targets={ Target.from_dict(target_dict=target_dict) diff --git a/src/mock_vws/target.py b/src/mock_vws/target.py index afd8f20b1..dbbbb35d7 100644 --- a/src/mock_vws/target.py +++ b/src/mock_vws/target.py @@ -29,6 +29,7 @@ class TargetDict(TypedDict): processing_time_seconds: float application_metadata: str | None target_id: str + target_type: str last_modified_date: str delete_date_optional: str | None upload_date: str @@ -69,6 +70,7 @@ class Target: previous_month_recos: int = 0 reco_rating: str = "" target_id: str = field(default_factory=_random_hex) + target_type: str = "cloud_target" total_recos: int = 0 upload_date: datetime.datetime = field(default_factory=_time_now) @@ -156,6 +158,7 @@ def from_dict(cls, target_dict: TargetDict) -> Self: processing_time_seconds = target_dict["processing_time_seconds"] application_metadata = target_dict["application_metadata"] target_id = target_dict["target_id"] + target_type = target_dict["target_type"] delete_date_optional = target_dict["delete_date_optional"] if delete_date_optional is None: delete_date = None @@ -175,6 +178,7 @@ def from_dict(cls, target_dict: TargetDict) -> Self: ) return cls( target_id=target_id, + target_type=target_type, name=name, active_flag=active_flag, width=width, @@ -203,6 +207,7 @@ def to_dict(self) -> TargetDict: "processing_time_seconds": float(self.processing_time_seconds), "application_metadata": self.application_metadata, "target_id": self.target_id, + "target_type": self.target_type, "last_modified_date": self.last_modified_date.isoformat(), "delete_date_optional": delete_date, "upload_date": self.upload_date.isoformat(), diff --git a/tests/conftest.py b/tests/conftest.py index 64ef1427f..2a81d82f6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,7 @@ import base64 import binascii +import dataclasses import io import uuid @@ -74,6 +75,15 @@ def target_id( ) +@pytest.fixture +def vucloud_database(vuforia_database: VuforiaDatabase) -> VuforiaDatabase: + """A database configured to create VuMark targets by default.""" + return dataclasses.replace( + vuforia_database, + default_target_type="vumark", + ) + + @pytest.fixture( params=[ "add_target", diff --git a/tests/mock_vws/fixtures/vuforia_backends.py b/tests/mock_vws/fixtures/vuforia_backends.py index 39013b5b9..f599d96fc 100644 --- a/tests/mock_vws/fixtures/vuforia_backends.py +++ b/tests/mock_vws/fixtures/vuforia_backends.py @@ -86,6 +86,7 @@ def _enable_use_mock_vuforia( server_secret_key=working_database.server_secret_key, client_access_key=working_database.client_access_key, client_secret_key=working_database.client_secret_key, + default_target_type=working_database.default_target_type, ) inactive_database = VuforiaDatabase( @@ -95,6 +96,7 @@ def _enable_use_mock_vuforia( server_secret_key=inactive_database.server_secret_key, client_access_key=inactive_database.client_access_key, client_secret_key=inactive_database.client_secret_key, + default_target_type=inactive_database.default_target_type, ) with MockVWS() as mock: diff --git a/tests/mock_vws/test_invalid_given_id.py b/tests/mock_vws/test_invalid_given_id.py index 067867f75..1fd1739fa 100644 --- a/tests/mock_vws/test_invalid_given_id.py +++ b/tests/mock_vws/test_invalid_given_id.py @@ -4,7 +4,10 @@ be given. """ +import io from http import HTTPStatus +from types import SimpleNamespace +from unittest.mock import patch import pytest from vws import VWS @@ -49,3 +52,36 @@ def test_not_real_id( status_code=HTTPStatus.NOT_FOUND, result_code=ResultCodes.UNKNOWN_TARGET, ) + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestTargetIdNamedInstances: + """Regression tests for a target ID with value ``instances``.""" + + @staticmethod + def test_summary_path_handles_target_id_named_instances( + image_file_success_state_low_rating: io.BytesIO, + vws_client: VWS, + ) -> None: + """`/summary/{target_id}` should use the final path segment as + ID. + """ + target_id = "instances" + with patch( + "mock_vws.target.uuid.uuid4", + return_value=SimpleNamespace(hex=target_id), + ): + created_target_id = vws_client.add_target( + name="example_target", + width=1, + image=image_file_success_state_low_rating, + active_flag=True, + application_metadata=None, + ) + assert created_target_id == target_id + + vws_client.wait_for_target_processed(target_id=created_target_id) + report = vws_client.get_target_summary_report( + target_id=created_target_id, + ) + assert report.target_name == "example_target" diff --git a/tests/mock_vws/test_vumark_generation.py b/tests/mock_vws/test_vumark_generation.py new file mode 100644 index 000000000..f839d05c6 --- /dev/null +++ b/tests/mock_vws/test_vumark_generation.py @@ -0,0 +1,490 @@ +"""Tests for the VuMark Generation API endpoint.""" + +import dataclasses +import io +import json +import uuid +from http import HTTPMethod, HTTPStatus + +import pytest +import requests +from vws import VWS +from vws_auth_tools import authorization_header, rfc_1123_date + +from mock_vws._constants import ResultCodes +from mock_vws.database import VuforiaDatabase + +VWS_HOST = "https://vws.vuforia.com" + + +def _generate_vumark_instance( + *, + database: VuforiaDatabase, + target_id: str, + instance_id: str | int | None, + accept: str, +) -> requests.Response: + """Generate a VuMark instance. + + Args: + database: The Vuforia database credentials. + target_id: The target ID. + instance_id: The VuMark instance ID. + accept: The Accept header value. + + Returns: + The response from the API. + """ + request_path = f"/targets/{target_id}/instances" + content = json.dumps(obj={"instance_id": instance_id}).encode() + date = rfc_1123_date() + content_type = "application/json" + + authorization_string = authorization_header( + access_key=database.server_access_key, + secret_key=database.server_secret_key, + method=HTTPMethod.POST, + content=content, + content_type=content_type, + date=date, + request_path=request_path, + ) + + headers = { + "Authorization": authorization_string, + "Date": date, + "Content-Length": str(object=len(content)), + "Content-Type": content_type, + "Accept": accept, + } + + url = VWS_HOST + request_path + return requests.post(url=url, data=content, headers=headers, timeout=30) + + +def _generate_vumark_instance_with_body( + *, + database: VuforiaDatabase, + target_id: str, + body: dict[str, str | int | None], + accept: str, +) -> requests.Response: + """Generate a VuMark instance with a custom body. + + Args: + database: The Vuforia database credentials. + target_id: The target ID. + body: The request body. + accept: The Accept header value. + + Returns: + The response from the API. + """ + request_path = f"/targets/{target_id}/instances" + content = json.dumps(obj=body).encode() + date = rfc_1123_date() + content_type = "application/json" + + authorization_string = authorization_header( + access_key=database.server_access_key, + secret_key=database.server_secret_key, + method=HTTPMethod.POST, + content=content, + content_type=content_type, + date=date, + request_path=request_path, + ) + + headers = { + "Authorization": authorization_string, + "Date": date, + "Content-Length": str(object=len(content)), + "Content-Type": content_type, + "Accept": accept, + } + + url = VWS_HOST + request_path + return requests.post(url=url, data=content, headers=headers, timeout=30) + + +@pytest.mark.usefixtures("verify_mock_vuforia") +class TestSuccessfulGeneration: + """Tests for successful VuMark instance generation.""" + + @pytest.fixture + def vuforia_database( # pylint: disable=no-self-use + self, + vucloud_database: VuforiaDatabase, + ) -> VuforiaDatabase: + """Use a VuCloud database for VuMark generation tests.""" + return vucloud_database + + @staticmethod + def test_svg_generation( + image_file_success_state_low_rating: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """SVG images can be generated.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_success_state_low_rating, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test123", + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.OK + assert response.headers["Content-Type"] == "image/svg+xml" + # Verify the response contains valid SVG + assert b" None: + """PNG images can be generated.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_success_state_low_rating, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test456", + accept="image/png", + ) + + assert response.status_code == HTTPStatus.OK + assert response.headers["Content-Type"] == "image/png" + # Verify the response contains PNG magic bytes + assert response.content[:8] == b"\x89PNG\r\n\x1a\n" + + @staticmethod + def test_pdf_generation( + image_file_success_state_low_rating: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """PDF documents can be generated.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_success_state_low_rating, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test789", + accept="application/pdf", + ) + + assert response.status_code == HTTPStatus.OK + assert response.headers["Content-Type"] == "application/pdf" + # Verify the response contains PDF header + assert response.content.startswith(b"%PDF-") + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestInvalidAcceptHeader: + """Tests for invalid Accept headers.""" + + @staticmethod + @pytest.mark.parametrize( + argnames="accept_header", + argvalues=[ + "text/html", + "application/json", + "image/jpeg", + "", + "invalid", + ], + ids=[ + "text/html", + "application/json", + "image/jpeg (not supported)", + "empty", + "invalid", + ], + ) + def test_invalid_accept_header( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + accept_header: str, + ) -> None: + """An error is returned for invalid Accept headers.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test123", + accept=accept_header, + ) + + assert response.status_code == HTTPStatus.BAD_REQUEST + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_ACCEPT_HEADER.value + ) + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestInvalidInstanceId: + """Tests for invalid instance IDs.""" + + @staticmethod + @pytest.mark.parametrize( + argnames="instance_id", + argvalues=[ + "", + None, + ], + ids=[ + "empty string", + "None", + ], + ) + def test_invalid_instance_id( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + instance_id: str | None, + ) -> None: + """An error is returned for invalid instance IDs.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id=instance_id, + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_INSTANCE_ID.value + ) + + @staticmethod + def test_missing_instance_id( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """An error is returned when instance_id is missing from the body. + + The key validator rejects the request with a "Fail" error (400) + because instance_id is a required field. + """ + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + + response = _generate_vumark_instance_with_body( + database=vuforia_database, + target_id=target_id, + body={}, # No instance_id + accept="image/svg+xml", + ) + + # Missing required keys return a "Fail" error + assert response.status_code == HTTPStatus.BAD_REQUEST + response_json = response.json() + assert response_json["result_code"] == ResultCodes.FAIL.value + + @staticmethod + def test_integer_instance_id( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """An error is returned when instance_id is not a string.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id=12345, + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_INSTANCE_ID.value + ) + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestInvalidTargetType: + """Tests for non-vumark target types.""" + + @staticmethod + def test_cloud_target( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """An error is returned for non-vumark targets.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test123", + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_TARGET_TYPE.value + ) + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestTargetStatusNotSuccess: + """Tests for targets not in success status.""" + + @pytest.fixture + def vuforia_database( # pylint: disable=no-self-use + self, + vuforia_database: VuforiaDatabase, + ) -> VuforiaDatabase: + """Override to create a VuMark database.""" + return dataclasses.replace( + vuforia_database, + default_target_type="vumark", + ) + + @staticmethod + def test_failed_target( + image_file_failed_state: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """An error is returned when target status is not success.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test123", + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.FORBIDDEN + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.TARGET_STATUS_NOT_SUCCESS.value + ) + + +@pytest.mark.usefixtures("mock_only_vuforia") +class TestResponseHeaders: + """Tests for response headers.""" + + @pytest.fixture + def vuforia_database( # pylint: disable=no-self-use + self, + vuforia_database: VuforiaDatabase, + ) -> VuforiaDatabase: + """Override to create a VuMark database.""" + return dataclasses.replace( + vuforia_database, + default_target_type="vumark", + ) + + @staticmethod + def test_response_headers( + image_file_success_state_low_rating: io.BytesIO, + vuforia_database: VuforiaDatabase, + vws_client: VWS, + ) -> None: + """The response includes expected headers.""" + target_id = vws_client.add_target( + name=uuid.uuid4().hex, + width=1, + image=image_file_success_state_low_rating, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + + response = _generate_vumark_instance( + database=vuforia_database, + target_id=target_id, + instance_id="test123", + accept="image/svg+xml", + ) + + assert response.status_code == HTTPStatus.OK + assert response.headers["Connection"] == "keep-alive" + assert response.headers["server"] == "envoy" + assert response.headers["Content-Type"] == "image/svg+xml" + assert "Content-Length" in response.headers + assert "Date" in response.headers