diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 6146edbec..86542874d 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -4,6 +4,10 @@ Changelog Next ---- +- Add ``VuMarkTarget`` class for VuMark template targets, alongside the renamed ``ImageTarget`` class (previously ``Target``). + ``ImageTarget`` is for image-based targets and ``VuMarkTarget`` is for VuMark template targets. + Both can be stored in a ``VuforiaDatabase``. + 2026.02.18.2 ------------ diff --git a/docs/source/mock-api-reference.rst b/docs/source/mock-api-reference.rst index 338855d2c..1b2ea255a 100644 --- a/docs/source/mock-api-reference.rst +++ b/docs/source/mock-api-reference.rst @@ -29,6 +29,10 @@ API Reference :members: :undoc-members: +.. autoenum:: mock_vws.database_type.DatabaseType + :members: + :undoc-members: + .. autoclass:: mock_vws.target.ImageTarget .. autoclass:: mock_vws.target.VuMarkTarget diff --git a/pyproject.toml b/pyproject.toml index c5a2f5685..df251f8d2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -452,6 +452,7 @@ ignore_names = [ # Used in TYPE_CHECKING for type hints "CloudDatabaseDict", "VuMarkDatabaseDict", + "VuMarkTargetDict", ] # Duplicate some of .gitignore exclude = [ ".venv" ] diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index 365309073..0b053e3bb 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -3,6 +3,7 @@ MPixel MiB MissingSchema Ubuntu +VuMark admin another's api diff --git a/src/mock_vws/_constants.py b/src/mock_vws/_constants.py index 52c88bb9b..c6077c62a 100644 --- a/src/mock_vws/_constants.py +++ b/src/mock_vws/_constants.py @@ -63,6 +63,7 @@ class ResultCodes(Enum): INVALID_ACCEPT_HEADER = "InvalidAcceptHeader" INVALID_INSTANCE_ID = "InvalidInstanceId" BAD_REQUEST = "BadRequest" + 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 1a308e4dd..3c5b6fd51 100644 --- a/src/mock_vws/_flask_server/target_manager.py +++ b/src/mock_vws/_flask_server/target_manager.py @@ -13,6 +13,7 @@ from pydantic_settings import BaseSettings from mock_vws.database import CloudDatabase, VuMarkDatabase +from mock_vws.database_type import DatabaseType from mock_vws.states import States from mock_vws.target import ImageTarget, VuMarkTarget from mock_vws.target_manager import TargetManager @@ -204,8 +205,13 @@ def create_cloud_database() -> Response: "state_name", random_database.state.name, ) + database_type_name = request_json.get( + "database_type_name", + random_database.database_type.name, + ) state = States[state_name] + database_type = DatabaseType[database_type_name] database = CloudDatabase( server_access_key=server_access_key, @@ -214,6 +220,7 @@ def create_cloud_database() -> Response: client_secret_key=client_secret_key, database_name=database_name, state=state, + database_type=database_type, ) try: TARGET_MANAGER.add_cloud_database(cloud_database=database) @@ -283,11 +290,10 @@ def create_target(database_name: str) -> Response: if database.database_name == database_name ) request_json = json.loads(s=request.data) - image_base64 = request_json["image_base64"] - image_bytes = base64.b64decode(s=image_base64) settings = TargetManagerSettings.model_validate(obj={}) - target_tracking_rater = settings.target_rater.to_target_rater() + image_bytes = base64.b64decode(s=request_json["image_base64"]) + target_tracking_rater = settings.target_rater.to_target_rater() target = ImageTarget( name=request_json["name"], width=request_json["width"], @@ -343,7 +349,7 @@ def delete_target(database_name: str, target_id: str) -> Response: target = database.get_target(target_id=target_id) now = datetime.datetime.now(tz=target.upload_date.tzinfo) # See https://github.com/facebook/pyrefly/issues/1897 - new_target = copy.replace( + new_target: ImageTarget = copy.replace( target, # pyrefly: ignore[bad-argument-type] delete_date=now, ) @@ -369,24 +375,22 @@ def update_target(database_name: str, target_id: str) -> Response: target = database.get_target(target_id=target_id) request_json = json.loads(s=request.data) - width = request_json.get("width", target.width) name = request_json.get("name", target.name) active_flag = request_json.get("active_flag", target.active_flag) + + gmt = ZoneInfo(key="GMT") + last_modified_date = datetime.datetime.now(tz=gmt) + + width = request_json.get("width", target.width) application_metadata = request_json.get( "application_metadata", target.application_metadata, ) - image_value = target.image_value - request_json = json.loads(s=request.data) if "image" in request_json: image_value = base64.b64decode(s=request_json["image"]) - - gmt = ZoneInfo(key="GMT") - last_modified_date = datetime.datetime.now(tz=gmt) - # See https://github.com/facebook/pyrefly/issues/1897 - new_target = copy.replace( + new_target: ImageTarget = copy.replace( target, # pyrefly: ignore[bad-argument-type] name=name, width=width, diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index 1c98d1336..fc9bb8a84 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -32,6 +32,7 @@ FailError, InvalidAcceptHeaderError, InvalidInstanceIdError, + InvalidTargetTypeError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, @@ -278,13 +279,16 @@ def get_target(target_id: str) -> Response: target for target in database.targets if target.target_id == target_id ) + width = target.width + tracking_rating = target.tracking_rating + reco_rating = target.reco_rating target_record = { "target_id": target.target_id, "active_flag": target.active_flag, "name": target.name, - "width": target.width, - "tracking_rating": target.tracking_rating, - "reco_rating": target.reco_rating, + "width": width, + "tracking_rating": tracking_rating, + "reco_rating": reco_rating, } date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True) @@ -394,6 +398,16 @@ def generate_vumark_instance(target_id: str) -> Response: # ``target_id`` is validated by request validators. del target_id + database = get_database_matching_server_keys( + request_headers=dict(request.headers), + request_body=request.data, + request_method=request.method, + request_path=request.path, + databases=all_databases, + ) + if not isinstance(database, VuMarkDatabase): + raise InvalidTargetTypeError + accept = request.headers.get(key="Accept", default="") valid_accept_types: dict[str, bytes] = { "image/png": VUMARK_PNG, @@ -503,6 +517,10 @@ def target_summary(target_id: str) -> Response: (target,) = ( target for target in database.targets if target.target_id == target_id ) + tracking_rating = target.tracking_rating + total_recos = target.total_recos + current_month_recos = target.current_month_recos + previous_month_recos = target.previous_month_recos body = { "status": target.status, "transaction_id": uuid.uuid4().hex, @@ -511,10 +529,10 @@ def target_summary(target_id: str) -> Response: "target_name": target.name, "upload_date": target.upload_date.strftime(format="%Y-%m-%d"), "active_flag": target.active_flag, - "tracking_rating": target.tracking_rating, - "total_recos": target.total_recos, - "current_month_recos": target.current_month_recos, - "previous_month_recos": target.previous_month_recos, + "tracking_rating": tracking_rating, + "total_recos": total_recos, + "current_month_recos": current_month_recos, + "previous_month_recos": previous_month_recos, } date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True) 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 5a5b7da96..deac884bf 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 @@ -32,17 +32,19 @@ FailError, InvalidAcceptHeaderError, InvalidInstanceIdError, + InvalidTargetTypeError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, ) +from mock_vws.database import VuMarkDatabase from mock_vws.image_matchers import ImageMatcher from mock_vws.target import ImageTarget from mock_vws.target_manager import TargetManager from mock_vws.target_raters import TargetTrackingRater if TYPE_CHECKING: - from mock_vws.database import CloudDatabase, VuMarkDatabase + from mock_vws.database import CloudDatabase _TARGET_ID_PATTERN = "[A-Za-z0-9]+" @@ -268,7 +270,7 @@ def delete_target(self, request: PreparedRequest) -> _ResponseType: now = datetime.datetime.now(tz=target.upload_date.tzinfo) # See https://github.com/facebook/pyrefly/issues/1897 - new_target = copy.replace( + new_target: ImageTarget = copy.replace( target, # pyrefly: ignore[bad-argument-type] delete_date=now, ) @@ -324,6 +326,16 @@ def generate_vumark_instance( databases=all_databases, ) + 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=all_databases, + ) + if not isinstance(database, VuMarkDatabase): + raise InvalidTargetTypeError + accept = dict(request.headers).get("Accept", "") if accept not in valid_accept_types: raise InvalidAcceptHeaderError @@ -500,13 +512,16 @@ def get_target(self, request: PreparedRequest) -> _ResponseType: target_id = request.path_url.split(sep="/")[-1] target = database.get_target(target_id=target_id) + width = target.width + tracking_rating = target.tracking_rating + reco_rating = target.reco_rating target_record = { "target_id": target.target_id, "active_flag": target.active_flag, "name": target.name, - "width": target.width, - "tracking_rating": target.tracking_rating, - "reco_rating": target.reco_rating, + "width": width, + "tracking_rating": tracking_rating, + "reco_rating": reco_rating, } date = email.utils.formatdate( timeval=None, @@ -653,17 +668,8 @@ def update_target(self, request: PreparedRequest) -> _ResponseType: ) request_json: dict[str, Any] = json.loads(s=request.body or b"") - width = request_json.get("width", target.width) name = request_json.get("name", target.name) active_flag = request_json.get("active_flag", target.active_flag) - application_metadata = request_json.get( - "application_metadata", - target.application_metadata, - ) - - image_value = target.image_value - if "image" in request_json: - image_value = base64.b64decode(s=request_json["image"]) if "active_flag" in request_json and active_flag is None: fail_exception = FailError(status_code=HTTPStatus.BAD_REQUEST) @@ -673,6 +679,19 @@ def update_target(self, request: PreparedRequest) -> _ResponseType: fail_exception.response_text, ) + gmt = ZoneInfo(key="GMT") + last_modified_date = datetime.datetime.now(tz=gmt) + + width = request_json.get("width", target.width) + application_metadata = request_json.get( + "application_metadata", + target.application_metadata, + ) + + image_value = target.image_value + if "image" in request_json: + image_value = base64.b64decode(s=request_json["image"]) + if ( "application_metadata" in request_json and application_metadata is None @@ -684,11 +703,8 @@ def update_target(self, request: PreparedRequest) -> _ResponseType: fail_exception.response_text, ) - gmt = ZoneInfo(key="GMT") - last_modified_date = datetime.datetime.now(tz=gmt) - # See https://github.com/facebook/pyrefly/issues/1897 - new_target = copy.replace( + new_target: ImageTarget = copy.replace( target, # pyrefly: ignore[bad-argument-type] name=name, width=width, @@ -755,6 +771,10 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType: localtime=False, usegmt=True, ) + tracking_rating = target.tracking_rating + total_recos = target.total_recos + current_month_recos = target.current_month_recos + previous_month_recos = target.previous_month_recos body = { "status": target.status, "transaction_id": uuid.uuid4().hex, @@ -763,10 +783,10 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType: "target_name": target.name, "upload_date": target.upload_date.strftime(format="%Y-%m-%d"), "active_flag": target.active_flag, - "tracking_rating": target.tracking_rating, - "total_recos": target.total_recos, - "current_month_recos": target.current_month_recos, - "previous_month_recos": target.previous_month_recos, + "tracking_rating": tracking_rating, + "total_recos": total_recos, + "current_month_recos": current_month_recos, + "previous_month_recos": previous_month_recos, } body_json = json_dump(body=body) headers = { diff --git a/src/mock_vws/_services_validators/exceptions.py b/src/mock_vws/_services_validators/exceptions.py index a722f29ab..da058422d 100644 --- a/src/mock_vws/_services_validators/exceptions.py +++ b/src/mock_vws/_services_validators/exceptions.py @@ -646,6 +646,46 @@ def __init__(self) -> None: } +@beartype +class InvalidTargetTypeError(ValidatorError): + """Exception raised when the target type is not valid for the + operation. + """ + + 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 TargetStatusProcessingError(ValidatorError): """Exception raised when trying to delete a target which is processing.""" diff --git a/src/mock_vws/database.py b/src/mock_vws/database.py index b030a9e4e..d4e2389a8 100644 --- a/src/mock_vws/database.py +++ b/src/mock_vws/database.py @@ -8,6 +8,7 @@ from beartype import beartype from mock_vws._constants import TargetStatuses +from mock_vws.database_type import DatabaseType from mock_vws.states import States from mock_vws.target import ( ImageTarget, @@ -27,6 +28,7 @@ class CloudDatabaseDict(TypedDict): client_access_key: str client_secret_key: str state_name: str + database_type_name: str targets: Iterable[ImageTargetDict] @@ -81,6 +83,7 @@ class CloudDatabase: hash=False, ) state: States = States.WORKING + database_type: DatabaseType = DatabaseType.CLOUD_RECO request_quota: int = 100000 reco_threshold: int = 1000 @@ -91,7 +94,9 @@ class CloudDatabase: def to_dict(self) -> CloudDatabaseDict: """Dump a target to a dictionary which can be loaded as JSON.""" - targets = [target.to_dict() for target in self.targets] + targets: list[ImageTargetDict] = [ + target.to_dict() for target in self.targets + ] return { "database_name": self.database_name, "server_access_key": self.server_access_key, @@ -99,6 +104,7 @@ def to_dict(self) -> CloudDatabaseDict: "client_access_key": self.client_access_key, "client_secret_key": self.client_secret_key, "state_name": self.state.name, + "database_type_name": self.database_type.name, "targets": targets, } @@ -112,6 +118,11 @@ def get_target(self, target_id: str) -> ImageTarget: @classmethod def from_dict(cls, database_dict: CloudDatabaseDict) -> Self: """Load a database from a dictionary.""" + targets: set[ImageTarget] = { + ImageTarget.from_dict(target_dict=target_dict) + for target_dict in database_dict["targets"] + } + return cls( database_name=database_dict["database_name"], server_access_key=database_dict["server_access_key"], @@ -119,10 +130,8 @@ def from_dict(cls, database_dict: CloudDatabaseDict) -> Self: client_access_key=database_dict["client_access_key"], client_secret_key=database_dict["client_secret_key"], state=States[database_dict["state_name"]], - targets={ - ImageTarget.from_dict(target_dict=target_dict) - for target_dict in database_dict["targets"] - }, + database_type=DatabaseType[database_dict["database_type_name"]], + targets=targets, ) @property diff --git a/src/mock_vws/database_type.py b/src/mock_vws/database_type.py new file mode 100644 index 000000000..bd72733d4 --- /dev/null +++ b/src/mock_vws/database_type.py @@ -0,0 +1,13 @@ +"""Vuforia database types.""" + +from enum import StrEnum, auto, unique + +from beartype import beartype + + +@beartype +@unique +class DatabaseType(StrEnum): + """Constants representing various database types.""" + + CLOUD_RECO = auto() diff --git a/src/mock_vws/target.py b/src/mock_vws/target.py index d381884d8..50e9ec034 100644 --- a/src/mock_vws/target.py +++ b/src/mock_vws/target.py @@ -27,7 +27,7 @@ class VuMarkTargetDict(TypedDict): class ImageTargetDict(TypedDict): - """A dictionary type which represents a target.""" + """A dictionary type which represents an image target.""" name: str width: float @@ -58,8 +58,7 @@ def _time_now() -> datetime.datetime: @beartype(conf=BeartypeConf(is_pep484_tower=True)) @dataclass(frozen=True, eq=True) class ImageTarget: - """ - A Vuforia Target as managed in + """A Vuforia image target as managed in https://developer.vuforia.com/target-manager. """ diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index cc27c8bf7..8fa5d91de 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -395,6 +395,7 @@ def test_to_dict(high_quality_image: io.BytesIO) -> None: assert len(database.targets) == 1 target = next(iter(database.targets)) + assert isinstance(target, ImageTarget) target_dict = target.to_dict() # The dictionary is JSON dump-able @@ -431,6 +432,7 @@ def test_to_dict_deleted(high_quality_image: io.BytesIO) -> None: assert len(database.targets) == 1 target = next(iter(database.targets)) + assert isinstance(target, ImageTarget) target_dict = target.to_dict() # The dictionary is JSON dump-able diff --git a/tests/mock_vws/test_vumark_generation_api.py b/tests/mock_vws/test_vumark_generation_api.py index 219c27440..2258988a6 100644 --- a/tests/mock_vws/test_vumark_generation_api.py +++ b/tests/mock_vws/test_vumark_generation_api.py @@ -6,10 +6,13 @@ 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 CloudDatabase from tests.mock_vws.fixtures.credentials import VuMarkCloudDatabase +from tests.mock_vws.utils import make_image_file _VWS_HOST = "https://vws.vuforia.com" _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" @@ -19,22 +22,24 @@ def _make_vumark_request( *, - vumark_vuforia_database: VuMarkCloudDatabase, + server_access_key: str, + server_secret_key: str, + target_id: str, 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" + request_path = f"/targets/{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, + access_key=server_access_key, + secret_key=server_secret_key, method=HTTPMethod.POST, content=content, content_type=content_type, @@ -87,7 +92,9 @@ def test_generate_instance_format( ) -> None: """A VuMark instance can be generated in the requested format.""" response = _make_vumark_request( - vumark_vuforia_database=vumark_vuforia_database, + server_access_key=vumark_vuforia_database.server_access_key, + server_secret_key=vumark_vuforia_database.server_secret_key, + target_id=vumark_vuforia_database.target_id, instance_id=uuid4().hex, accept=accept, ) @@ -106,7 +113,9 @@ def test_invalid_accept_header( ) -> None: """An unsupported Accept header returns an error.""" response = _make_vumark_request( - vumark_vuforia_database=vumark_vuforia_database, + server_access_key=vumark_vuforia_database.server_access_key, + server_secret_key=vumark_vuforia_database.server_secret_key, + target_id=vumark_vuforia_database.target_id, instance_id=uuid4().hex, accept="text/plain", ) @@ -124,7 +133,9 @@ def test_empty_instance_id( ) -> None: """An empty instance_id returns InvalidInstanceId.""" response = _make_vumark_request( - vumark_vuforia_database=vumark_vuforia_database, + server_access_key=vumark_vuforia_database.server_access_key, + server_secret_key=vumark_vuforia_database.server_secret_key, + target_id=vumark_vuforia_database.target_id, instance_id="", accept="image/png", ) @@ -135,3 +146,41 @@ def test_empty_instance_id( response_json["result_code"] == ResultCodes.INVALID_INSTANCE_ID.value ) + + @staticmethod + def test_non_vumark_database( + vuforia_database: CloudDatabase, + ) -> None: + """Generating a VuMark instance for a target in a non-VuMark + database returns InvalidTargetType. + """ + vws_client = VWS( + server_access_key=vuforia_database.server_access_key, + server_secret_key=vuforia_database.server_secret_key, + ) + image = make_image_file( + file_format="PNG", + color_space="RGB", + width=8, + height=8, + ) + target_id = vws_client.add_target( + name="test", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) + response = _make_vumark_request( + server_access_key=vuforia_database.server_access_key, + server_secret_key=vuforia_database.server_secret_key, + target_id=target_id, + instance_id=uuid4().hex, + accept="image/png", + ) + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.INVALID_TARGET_TYPE.value + )