diff --git a/src/mock_vws/_constants.py b/src/mock_vws/_constants.py index d9af993b5..68bf9375f 100644 --- a/src/mock_vws/_constants.py +++ b/src/mock_vws/_constants.py @@ -11,6 +11,21 @@ b"\xaeB`\x82" ) +VUMARK_SVG = ( + b'' +) + +VUMARK_PDF = ( + b"%PDF-1.4\n" + b"1 0 obj<>endobj\n" + b"2 0 obj<>endobj\n" + b"3 0 obj<>endobj\n" + b"xref\n0 4\n" + b"0000000000 65535 f \n" + b"trailer<>\n" + b"startxref\n9\n%%EOF" +) + @beartype @unique @@ -45,6 +60,8 @@ class ResultCodes(Enum): PROJECT_INACTIVE = "ProjectInactive" INACTIVE_PROJECT = "InactiveProject" TOO_MANY_REQUESTS = "TooManyRequests" + INVALID_ACCEPT_HEADER = "InvalidAcceptHeader" + INVALID_INSTANCE_ID = "InvalidInstanceId" @beartype diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index 99b8c3b8e..664571b6f 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -18,12 +18,20 @@ from flask import Flask, Response, request from pydantic_settings import BaseSettings -from mock_vws._constants import VUMARK_PNG, ResultCodes, TargetStatuses +from mock_vws._constants import ( + VUMARK_PDF, + VUMARK_PNG, + VUMARK_SVG, + ResultCodes, + TargetStatuses, +) from mock_vws._database_matchers import get_database_matching_server_keys from mock_vws._mock_common import json_dump from mock_vws._services_validators import run_services_validators from mock_vws._services_validators.exceptions import ( FailError, + InvalidAcceptHeaderError, + InvalidInstanceIdError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, @@ -351,10 +359,27 @@ def generate_vumark_instance(target_id: str) -> Response: """ # ``target_id`` is validated by request validators. del target_id + + accept = request.headers.get(key="Accept", default="") + valid_accept_types: dict[str, bytes] = { + "image/png": VUMARK_PNG, + "image/svg+xml": VUMARK_SVG, + "application/pdf": VUMARK_PDF, + } + if accept not in valid_accept_types: + raise InvalidAcceptHeaderError + + request_json = json.loads(s=request.data) + instance_id = request_json.get("instance_id", "") + if not instance_id: + raise InvalidInstanceIdError + + response_body = valid_accept_types[accept] + content_type = accept date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True) headers = { "Connection": "keep-alive", - "Content-Type": "image/png", + "Content-Type": content_type, "server": "envoy", "Date": date, "x-envoy-upstream-service-time": "5", @@ -364,7 +389,7 @@ def generate_vumark_instance(target_id: str) -> Response: } return Response( status=HTTPStatus.OK, - response=VUMARK_PNG, + response=response_body, headers=headers, ) 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 303d4a2b4..320b776ef 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 @@ -18,12 +18,20 @@ from beartype import BeartypeConf, beartype from requests.models import PreparedRequest -from mock_vws._constants import VUMARK_PNG, ResultCodes, TargetStatuses +from mock_vws._constants import ( + VUMARK_PDF, + VUMARK_PNG, + VUMARK_SVG, + ResultCodes, + TargetStatuses, +) from mock_vws._database_matchers import get_database_matching_server_keys from mock_vws._mock_common import Route, json_dump from mock_vws._services_validators import run_services_validators from mock_vws._services_validators.exceptions import ( FailError, + InvalidAcceptHeaderError, + InvalidInstanceIdError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, @@ -295,14 +303,33 @@ def generate_vumark_instance( self, request: PreparedRequest ) -> _ResponseType: """Generate a VuMark instance.""" - 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, - ) + valid_accept_types: dict[str, bytes] = { + "image/png": VUMARK_PNG, + "image/svg+xml": VUMARK_SVG, + "application/pdf": VUMARK_PDF, + } + 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 = dict(request.headers).get("Accept", "") + if accept not in valid_accept_types: + raise InvalidAcceptHeaderError + + request_json = json.loads(s=_body_bytes(request=request)) + instance_id = request_json.get("instance_id", "") + if not instance_id: + raise InvalidInstanceIdError + except ValidatorError as exc: + return exc.status_code, exc.headers, exc.response_text + response_body = valid_accept_types[accept] + content_type = accept date = email.utils.formatdate( timeval=None, localtime=False, @@ -310,7 +337,7 @@ def generate_vumark_instance( ) headers = { "Connection": "keep-alive", - "Content-Type": "image/png", + "Content-Type": content_type, "Date": date, "server": "envoy", "x-envoy-upstream-service-time": "5", @@ -318,7 +345,7 @@ def generate_vumark_instance( "x-aws-region": "us-east-2, us-west-2", "x-content-type-options": "nosniff", } - return HTTPStatus.OK, headers, VUMARK_PNG + return HTTPStatus.OK, headers, response_body @route(path_pattern="/summary", http_methods={HTTPMethod.GET}) def database_summary(self, request: PreparedRequest) -> _ResponseType: diff --git a/src/mock_vws/_services_validators/exceptions.py b/src/mock_vws/_services_validators/exceptions.py index 4bbc5dab8..f7cbe2439 100644 --- a/src/mock_vws/_services_validators/exceptions.py +++ b/src/mock_vws/_services_validators/exceptions.py @@ -530,6 +530,82 @@ def __init__(self) -> None: } +@beartype +class InvalidAcceptHeaderError(ValidatorError): + """Exception raised when an unsupported Accept header 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.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", + } + + +@beartype +class InvalidInstanceIdError(ValidatorError): + """Exception raised when an invalid 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 TargetStatusProcessingError(ValidatorError): """Exception raised when trying to delete a target which is processing.""" diff --git a/tests/mock_vws/test_vumark_generation_api.py b/tests/mock_vws/test_vumark_generation_api.py index f2ec64527..2b589ebd6 100644 --- a/tests/mock_vws/test_vumark_generation_api.py +++ b/tests/mock_vws/test_vumark_generation_api.py @@ -8,53 +8,130 @@ import requests from vws_auth_tools import authorization_header, rfc_1123_date +from mock_vws._constants import ResultCodes from tests.mock_vws.fixtures.credentials import VuMarkVuforiaDatabase _VWS_HOST = "https://vws.vuforia.com" _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" +_PDF_SIGNATURE = b"%PDF" +_SVG_START = b"<" + + +def _make_vumark_request( + *, + vumark_vuforia_database: VuMarkVuforiaDatabase, + instance_id: str, + accept: str, +) -> requests.Response: + """Send a VuMark instance generation request and return the + response. + """ + request_path = f"/targets/{vumark_vuforia_database.target_id}/instances" + content_type = "application/json" + content = json.dumps(obj={"instance_id": instance_id}).encode( + encoding="utf-8" + ) + date = rfc_1123_date() + authorization_string = authorization_header( + access_key=vumark_vuforia_database.server_access_key, + secret_key=vumark_vuforia_database.server_secret_key, + method=HTTPMethod.POST, + content=content, + content_type=content_type, + date=date, + request_path=request_path, + ) + + return requests.post( + url=_VWS_HOST + request_path, + headers={ + "Accept": accept, + "Authorization": authorization_string, + "Content-Length": str(object=len(content)), + "Content-Type": content_type, + "Date": date, + }, + data=content, + timeout=30, + ) @pytest.mark.usefixtures("verify_mock_vuforia") class TestGenerateInstance: """Tests for the VuMark instance generation endpoint.""" + @pytest.mark.parametrize( + argnames=("accept", "expected_content_type", "expected_signature"), + argvalues=[ + pytest.param("image/png", "image/png", _PNG_SIGNATURE, id="png"), + pytest.param( + "image/svg+xml", + "image/svg+xml", + _SVG_START, + id="svg", + ), + pytest.param( + "application/pdf", + "application/pdf", + _PDF_SIGNATURE, + id="pdf", + ), + ], + ) @staticmethod - def test_generate_instance_success( + def test_generate_instance_format( + accept: str, + expected_content_type: str, + expected_signature: bytes, vumark_vuforia_database: VuMarkVuforiaDatabase, ) -> None: - """A VuMark instance can be generated with valid template settings.""" - target_id = vumark_vuforia_database.target_id - request_path = f"/targets/{target_id}/instances" - content_type = "application/json" - generated_instance_id = uuid4().hex - body_dict = {"instance_id": generated_instance_id} - content = json.dumps(obj=body_dict).encode(encoding="utf-8") - date = rfc_1123_date() - authorization_string = authorization_header( - access_key=vumark_vuforia_database.server_access_key, - secret_key=vumark_vuforia_database.server_secret_key, - method=HTTPMethod.POST, - content=content, - content_type=content_type, - date=date, - request_path=request_path, + """A VuMark instance can be generated in the requested format.""" + response = _make_vumark_request( + vumark_vuforia_database=vumark_vuforia_database, + instance_id=uuid4().hex, + accept=accept, ) - response = requests.post( - url=_VWS_HOST + request_path, - headers={ - "Accept": "image/png", - "Authorization": authorization_string, - "Content-Length": str(object=len(content)), - "Content-Type": content_type, - "Date": date, - }, - data=content, - timeout=30, + assert response.status_code == HTTPStatus.OK + assert ( + response.headers["Content-Type"].split(sep=";")[0] + == expected_content_type ) + assert response.content.strip().startswith(expected_signature) + assert len(response.content) > len(expected_signature) - assert response.status_code == HTTPStatus.OK - content_type_value = response.headers["Content-Type"].split(sep=";")[0] - assert content_type_value == "image/png" - assert response.content.startswith(_PNG_SIGNATURE) - assert len(response.content) > len(_PNG_SIGNATURE) + @staticmethod + def test_invalid_accept_header( + vumark_vuforia_database: VuMarkVuforiaDatabase, + ) -> None: + """An unsupported Accept header returns an error.""" + response = _make_vumark_request( + vumark_vuforia_database=vumark_vuforia_database, + instance_id=uuid4().hex, + accept="text/plain", + ) + + assert response.status_code == HTTPStatus.BAD_REQUEST + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_ACCEPT_HEADER.value + ) + + @staticmethod + def test_empty_instance_id( + vumark_vuforia_database: VuMarkVuforiaDatabase, + ) -> None: + """An empty instance_id returns InvalidInstanceId.""" + response = _make_vumark_request( + vumark_vuforia_database=vumark_vuforia_database, + instance_id="", + accept="image/png", + ) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_INSTANCE_ID.value + )