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/admin/create_secrets_files.py b/admin/create_secrets_files.py index 13d2310d0..a9e3e69e4 100644 --- a/admin/create_secrets_files.py +++ b/admin/create_secrets_files.py @@ -19,11 +19,6 @@ from vws_web_tools import DatabaseDict, VuMarkDatabaseDict -VUMARK_TEMPLATE_SVG_FILE_PATH = Path(__file__).with_name( - name="vumark_template.svg", -) - - def _create_and_get_database_details( driver: "WebDriver", email_address: str, @@ -55,25 +50,6 @@ def _create_and_get_database_details( ) -def _create_and_get_vumark_details( - driver: "WebDriver", - vumark_database_name: str, -) -> "VuMarkDatabaseDict": - """Create a VuMark database and get its details. - - Returns VuMark database details. - """ - vws_web_tools.create_vumark_database( - driver=driver, - database_name=vumark_database_name, - ) - - return vws_web_tools.get_vumark_database_details( - driver=driver, - database_name=vumark_database_name, - ) - - def _generate_secrets_file_content( database_details: "DatabaseDict", vumark_details: "VuMarkDatabaseDict", @@ -103,27 +79,7 @@ def _generate_secrets_file_content( ) -def _create_and_get_vumark_target_id( - driver: "WebDriver", - vumark_database_name: str, - vumark_template_name: str, -) -> str: - """Upload a VuMark template and get its target ID.""" - vws_web_tools.upload_vumark_template( - driver=driver, - database_name=vumark_database_name, - svg_file_path=VUMARK_TEMPLATE_SVG_FILE_PATH, - template_name=vumark_template_name, - width=100.0, - ) - return vws_web_tools.get_vumark_target_id( - driver=driver, - database_name=vumark_database_name, - target_name=vumark_template_name, - ) - - -def _create_vuforia_resource_names() -> tuple[str, str, str, str]: +def _create_vuforia_resource_names() -> tuple[str, str]: """Create names for Vuforia resources.""" time = datetime.datetime.now(tz=datetime.UTC).strftime( format="%Y-%m-%d-%H-%M-%S", @@ -131,8 +87,6 @@ def _create_vuforia_resource_names() -> tuple[str, str, str, str]: return ( f"my-license-{time}", f"my-database-{time}", - f"my-vumark-database-{time}", - f"my-vumark-template-{time}", ) @@ -157,6 +111,18 @@ def main() -> None: "client_access_key": os.environ["INACTIVE_VUFORIA_CLIENT_ACCESS_KEY"], "client_secret_key": os.environ["INACTIVE_VUFORIA_CLIENT_SECRET_KEY"], } + vumark_details: VuMarkDatabaseDict = { + "database_name": os.environ[ + "VUMARK_VUFORIA_TARGET_MANAGER_DATABASE_NAME" + ], + "server_access_key": os.environ[ + "VUMARK_VUFORIA_SERVER_ACCESS_KEY" + ], + "server_secret_key": os.environ[ + "VUMARK_VUFORIA_SERVER_SECRET_KEY" + ], + } + vumark_target_id = os.environ["VUMARK_VUFORIA_TARGET_ID"] new_secrets_dir.mkdir(exist_ok=True) num_databases = 100 @@ -175,8 +141,6 @@ def main() -> None: ( license_name, database_name, - vumark_database_name, - vumark_template_name, ) = _create_vuforia_resource_names() try: @@ -195,33 +159,6 @@ def main() -> None: driver = None continue - try: - vumark_details = _create_and_get_vumark_details( - driver=driver, - vumark_database_name=vumark_database_name, - ) - except TimeoutException: - sys.stderr.write( - "Timed out waiting for VuMark setup/details after retries\n" - ) - driver.quit() - driver = None - continue - - try: - vumark_target_id = _create_and_get_vumark_target_id( - driver=driver, - vumark_database_name=vumark_database_name, - vumark_template_name=vumark_template_name, - ) - except TimeoutException: - sys.stderr.write( - "Timed out waiting for VuMark template upload after retries\n" - ) - driver.quit() - driver = None - continue - driver.quit() driver = None diff --git a/docs/source/mock-api-reference.rst b/docs/source/mock-api-reference.rst index ecb44f90b..f43708647 100644 --- a/docs/source/mock-api-reference.rst +++ b/docs/source/mock-api-reference.rst @@ -24,7 +24,13 @@ API Reference :members: :undoc-members: -.. autoclass:: mock_vws.target.Target +.. autoenum:: mock_vws.database_type.DatabaseType + :members: + :undoc-members: + +.. autoclass:: mock_vws.target.ImageTarget + +.. autoclass:: mock_vws.target.VuMarkTarget Image matchers -------------- diff --git a/pyproject.toml b/pyproject.toml index f4e49c096..e076b648c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -105,7 +105,7 @@ optional-dependencies.dev = [ "vulture==2.14", "vws-python==2026.2.15", "vws-test-fixtures==2023.3.5", - "vws-web-tools==2026.2.17.1", + "vws-web-tools==2026.2.20", "yamlfix==1.19.1", "zizmor==1.22.0", ] @@ -452,6 +452,7 @@ ignore_names = [ # Used in TYPE_CHECKING for type hints "DatabaseDict", "VuMarkDatabaseDict", + "VuMarkTargetDict", ] # Duplicate some of .gitignore exclude = [ ".venv" ] diff --git a/secrets.tar.gpg b/secrets.tar.gpg index a21cf43fd..32c9b9c6f 100644 Binary files a/secrets.tar.gpg and b/secrets.tar.gpg differ 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 e317549b1..68b72fe74 100644 --- a/src/mock_vws/_flask_server/target_manager.py +++ b/src/mock_vws/_flask_server/target_manager.py @@ -13,8 +13,9 @@ from pydantic_settings import BaseSettings from mock_vws.database import VuforiaDatabase +from mock_vws.database_type import DatabaseType from mock_vws.states import States -from mock_vws.target import Target +from mock_vws.target import ImageTarget, VuMarkTarget from mock_vws.target_manager import TargetManager from mock_vws.target_raters import ( BrisqueTargetTrackingRater, @@ -159,8 +160,13 @@ def create_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 = VuforiaDatabase( server_access_key=server_access_key, @@ -169,6 +175,7 @@ def create_database() -> Response: client_secret_key=client_secret_key, database_name=database_name, state=state, + database_type=database_type, ) try: TARGET_MANAGER.add_database(database=database) @@ -197,21 +204,29 @@ 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() - - target = Target( - name=request_json["name"], - width=request_json["width"], - image_value=image_bytes, - active_flag=request_json["active_flag"], - processing_time_seconds=request_json["processing_time_seconds"], - application_metadata=request_json["application_metadata"], - target_id=request_json["target_id"], - target_tracking_rater=target_tracking_rater, - ) + + target_type_name = request_json.get("target_type_name", "IMAGE") + if target_type_name == "VUMARK_TEMPLATE": + target: ImageTarget | VuMarkTarget = VuMarkTarget( + name=request_json["name"], + active_flag=request_json["active_flag"], + processing_time_seconds=request_json["processing_time_seconds"], + target_id=request_json["target_id"], + ) + else: + 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"], + image_value=image_bytes, + active_flag=request_json["active_flag"], + processing_time_seconds=request_json["processing_time_seconds"], + application_metadata=request_json["application_metadata"], + target_id=request_json["target_id"], + target_tracking_rater=target_tracking_rater, + ) database.targets.add(target) return Response( @@ -235,7 +250,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 | VuMarkTarget = copy.replace( # type: ignore[assignment] target, # pyrefly: ignore[bad-argument-type] delete_date=now, ) @@ -261,32 +276,39 @@ 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) - 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( - target, # pyrefly: ignore[bad-argument-type] - name=name, - width=width, - active_flag=active_flag, - application_metadata=application_metadata, - image_value=image_value, - last_modified_date=last_modified_date, - ) + if isinstance(target, ImageTarget): + 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"]) + # See https://github.com/facebook/pyrefly/issues/1897 + new_target: ImageTarget | VuMarkTarget = copy.replace( + target, # pyrefly: ignore[bad-argument-type] + name=name, + width=width, + active_flag=active_flag, + application_metadata=application_metadata, + image_value=image_value, + last_modified_date=last_modified_date, + ) + else: + # See https://github.com/facebook/pyrefly/issues/1897 + new_target = copy.replace( + target, # pyrefly: ignore[bad-argument-type] + name=name, + active_flag=active_flag, + last_modified_date=last_modified_date, + ) database.targets.remove(target) database.targets.add(new_target) diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index 664571b6f..936e5cf21 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -32,17 +32,19 @@ FailError, InvalidAcceptHeaderError, InvalidInstanceIdError, + InvalidTargetTypeError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, ) from mock_vws.database import VuforiaDatabase +from mock_vws.database_type import DatabaseType from mock_vws.image_matchers import ( ExactMatcher, ImageMatcher, StructuralSimilarityMatcher, ) -from mock_vws.target import Target +from mock_vws.target import ImageTarget from mock_vws.target_raters import ( HardcodedTargetTrackingRater, ) @@ -192,7 +194,7 @@ def add_target() -> Response: # This rater is not used. target_tracking_rater = HardcodedTargetTrackingRater(rating=1) - new_target = Target( + new_target = ImageTarget( name=name, width=request_json["width"], image_value=base64.b64decode(s=request_json["image"]), @@ -258,13 +260,21 @@ def get_target(target_id: str) -> Response: target for target in database.targets if target.target_id == target_id ) + if isinstance(target, ImageTarget): + width = target.width + tracking_rating = target.tracking_rating + reco_rating = target.reco_rating + else: + width = 0.0 + tracking_rating = -1 + 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) @@ -360,6 +370,17 @@ def generate_vumark_instance(target_id: str) -> Response: # ``target_id`` is validated by request validators. del target_id + 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, + ) + if database.database_type != DatabaseType.VUMARK: + raise InvalidTargetTypeError + accept = request.headers.get(key="Accept", default="") valid_accept_types: dict[str, bytes] = { "image/png": VUMARK_PNG, @@ -469,6 +490,16 @@ def target_summary(target_id: str) -> Response: (target,) = ( target for target in database.targets if target.target_id == target_id ) + if isinstance(target, ImageTarget): + tracking_rating = target.tracking_rating + total_recos = target.total_recos + current_month_recos = target.current_month_recos + previous_month_recos = target.previous_month_recos + else: + tracking_rating = -1 + total_recos = 0 + current_month_recos = 0 + previous_month_recos = 0 body = { "status": target.status, "transaction_id": uuid.uuid4().hex, @@ -477,10 +508,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 = { @@ -527,17 +558,21 @@ def get_duplicates(target_id: str) -> Response: ) other_targets = database.targets - {target} - similar_targets = [ - other.target_id - for other in other_targets - if image_match_checker( - first_image_content=target.image_value, - second_image_content=other.image_value, - ) - and TargetStatuses.FAILED.value not in {target.status, other.status} - and TargetStatuses.PROCESSING.value != other.status - and other.active_flag - ] + similar_targets = [] + if isinstance(target, ImageTarget): + similar_targets = [ + other.target_id + for other in other_targets + if isinstance(other, ImageTarget) + and image_match_checker( + first_image_content=target.image_value, + second_image_content=other.image_value, + ) + and TargetStatuses.FAILED.value + not in {target.status, other.status} + and TargetStatuses.PROCESSING.value != other.status + and other.active_flag + ] body = { "transaction_id": uuid.uuid4().hex, diff --git a/src/mock_vws/_query_tools.py b/src/mock_vws/_query_tools.py index b73f6c616..a929db38a 100644 --- a/src/mock_vws/_query_tools.py +++ b/src/mock_vws/_query_tools.py @@ -16,6 +16,7 @@ from mock_vws._mock_common import json_dump from mock_vws.database import VuforiaDatabase from mock_vws.image_matchers import ImageMatcher +from mock_vws.target import ImageTarget @beartype @@ -72,7 +73,8 @@ def get_query_match_response_text( matching_targets = [ target for target in database.targets - if query_match_checker( + if isinstance(target, ImageTarget) + and query_match_checker( first_image_content=target.image_value, second_image_content=image_value, ) 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 320b776ef..b9c7f193d 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,12 +32,14 @@ FailError, InvalidAcceptHeaderError, InvalidInstanceIdError, + InvalidTargetTypeError, TargetStatusNotSuccessError, TargetStatusProcessingError, ValidatorError, ) +from mock_vws.database_type import DatabaseType from mock_vws.image_matchers import ImageMatcher -from mock_vws.target import Target +from mock_vws.target import ImageTarget, VuMarkTarget from mock_vws.target_manager import TargetManager from mock_vws.target_raters import TargetTrackingRater @@ -187,7 +189,7 @@ def add_target(self, request: PreparedRequest) -> _ResponseType: application_metadata = request_json.get("application_metadata") - new_target = Target( + new_target = ImageTarget( name=request_json["name"], width=request_json["width"], image_value=base64.b64decode(s=request_json["image"]), @@ -265,7 +267,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 | VuMarkTarget = copy.replace( # type: ignore[assignment] target, # pyrefly: ignore[bad-argument-type] delete_date=now, ) @@ -317,6 +319,16 @@ def generate_vumark_instance( databases=self._target_manager.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=self._target_manager.databases, + ) + if database.database_type != DatabaseType.VUMARK: + raise InvalidTargetTypeError + accept = dict(request.headers).get("Accept", "") if accept not in valid_accept_types: raise InvalidAcceptHeaderError @@ -493,13 +505,21 @@ def get_target(self, request: PreparedRequest) -> _ResponseType: target_id = request.path_url.split(sep="/")[-1] target = database.get_target(target_id=target_id) + if isinstance(target, ImageTarget): + width = target.width + tracking_rating = target.tracking_rating + reco_rating = target.reco_rating + else: + width = 0.0 + tracking_rating = -1 + 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, @@ -561,18 +581,21 @@ def get_duplicates(self, request: PreparedRequest) -> _ResponseType: other_targets = database.targets - {target} - similar_targets = [ - other.target_id - for other in other_targets - if self._duplicate_match_checker( - first_image_content=target.image_value, - second_image_content=other.image_value, - ) - and TargetStatuses.FAILED.value - not in {target.status, other.status} - and TargetStatuses.PROCESSING.value != other.status - and other.active_flag - ] + similar_targets: list[str] = [] + if isinstance(target, ImageTarget): + similar_targets = [ + other.target_id + for other in other_targets + if isinstance(other, ImageTarget) + and self._duplicate_match_checker( + first_image_content=target.image_value, + second_image_content=other.image_value, + ) + and TargetStatuses.FAILED.value + not in {target.status, other.status} + and TargetStatuses.PROCESSING.value != other.status + and other.active_flag + ] date = email.utils.formatdate( timeval=None, @@ -646,17 +669,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) @@ -666,30 +680,49 @@ def update_target(self, request: PreparedRequest) -> _ResponseType: fail_exception.response_text, ) - if ( - "application_metadata" in request_json - and application_metadata is None - ): - fail_exception = FailError(status_code=HTTPStatus.BAD_REQUEST) - return ( - fail_exception.status_code, - fail_exception.headers, - 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( - target, # pyrefly: ignore[bad-argument-type] - name=name, - width=width, - active_flag=active_flag, - application_metadata=application_metadata, - image_value=image_value, - last_modified_date=last_modified_date, - ) + if isinstance(target, ImageTarget): + 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 + ): + fail_exception = FailError(status_code=HTTPStatus.BAD_REQUEST) + return ( + fail_exception.status_code, + fail_exception.headers, + fail_exception.response_text, + ) + + # See https://github.com/facebook/pyrefly/issues/1897 + new_target: ImageTarget | VuMarkTarget = copy.replace( + target, # pyrefly: ignore[bad-argument-type] + name=name, + width=width, + active_flag=active_flag, + application_metadata=application_metadata, + image_value=image_value, + last_modified_date=last_modified_date, + ) + else: + # See https://github.com/facebook/pyrefly/issues/1897 + new_target = copy.replace( + target, # pyrefly: ignore[bad-argument-type] + name=name, + active_flag=active_flag, + last_modified_date=last_modified_date, + ) database.targets.remove(target) database.targets.add(new_target) @@ -748,6 +781,16 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType: localtime=False, usegmt=True, ) + if isinstance(target, ImageTarget): + tracking_rating = target.tracking_rating + total_recos = target.total_recos + current_month_recos = target.current_month_recos + previous_month_recos = target.previous_month_recos + else: + tracking_rating = -1 + total_recos = 0 + current_month_recos = 0 + previous_month_recos = 0 body = { "status": target.status, "transaction_id": uuid.uuid4().hex, @@ -756,10 +799,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..e09a2785a 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.FORBIDDEN + 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 2e28a9f61..2a5a46af8 100644 --- a/src/mock_vws/database.py +++ b/src/mock_vws/database.py @@ -3,13 +3,19 @@ import uuid from collections.abc import Iterable from dataclasses import dataclass, field -from typing import Self, TypedDict +from typing import Self, TypedDict, cast 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 Target, TargetDict +from mock_vws.target import ( + ImageTarget, + ImageTargetDict, + VuMarkTarget, + VuMarkTargetDict, +) @beartype @@ -22,7 +28,8 @@ class DatabaseDict(TypedDict): client_access_key: str client_secret_key: str state_name: str - targets: Iterable[TargetDict] + database_type_name: str + targets: Iterable[ImageTargetDict | VuMarkTargetDict] @beartype @@ -61,8 +68,12 @@ class VuforiaDatabase: # ``frozen=True`` while still being able to keep the interface we want. # 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) + targets: set[ImageTarget | VuMarkTarget] = field( + default_factory=set[ImageTarget | VuMarkTarget], + hash=False, + ) state: States = States.WORKING + database_type: DatabaseType = DatabaseType.CLOUD_RECO request_quota: int = 100000 reco_threshold: int = 1000 @@ -73,7 +84,9 @@ class VuforiaDatabase: def to_dict(self) -> DatabaseDict: """Dump a target to a dictionary which can be loaded as JSON.""" - targets = [target.to_dict() for target in self.targets] + targets: list[ImageTargetDict | VuMarkTargetDict] = [ + target.to_dict() for target in self.targets + ] return { "database_name": self.database_name, "server_access_key": self.server_access_key, @@ -81,10 +94,11 @@ def to_dict(self) -> DatabaseDict: "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, } - def get_target(self, target_id: str) -> Target: + def get_target(self, target_id: str) -> ImageTarget | VuMarkTarget: """Return a target from the database with the given ID.""" (target,) = ( target for target in self.targets if target.target_id == target_id @@ -94,6 +108,21 @@ def get_target(self, target_id: str) -> Target: @classmethod def from_dict(cls, database_dict: DatabaseDict) -> Self: """Load a database from a dictionary.""" + targets: set[ImageTarget | VuMarkTarget] = set() + for target_dict in database_dict["targets"]: + if target_dict["target_type_name"] == "VUMARK_TEMPLATE": + targets.add( + VuMarkTarget.from_dict( + target_dict=target_dict, + ) + ) + else: + targets.add( + ImageTarget.from_dict( + target_dict=cast("ImageTargetDict", target_dict), + ) + ) + return cls( database_name=database_dict["database_name"], server_access_key=database_dict["server_access_key"], @@ -101,19 +130,17 @@ def from_dict(cls, database_dict: DatabaseDict) -> Self: client_access_key=database_dict["client_access_key"], client_secret_key=database_dict["client_secret_key"], state=States[database_dict["state_name"]], - targets={ - Target.from_dict(target_dict=target_dict) - for target_dict in database_dict["targets"] - }, + database_type=DatabaseType[database_dict["database_type_name"]], + targets=targets, ) @property - def not_deleted_targets(self) -> set[Target]: + def not_deleted_targets(self) -> set[ImageTarget | VuMarkTarget]: """All targets which have not been deleted.""" return {target for target in self.targets if not target.delete_date} @property - def active_targets(self) -> set[Target]: + def active_targets(self) -> set[ImageTarget | VuMarkTarget]: """All active targets.""" return { target @@ -123,7 +150,7 @@ def active_targets(self) -> set[Target]: } @property - def inactive_targets(self) -> set[Target]: + def inactive_targets(self) -> set[ImageTarget | VuMarkTarget]: """All inactive targets.""" return { target @@ -133,7 +160,7 @@ def inactive_targets(self) -> set[Target]: } @property - def failed_targets(self) -> set[Target]: + def failed_targets(self) -> set[ImageTarget | VuMarkTarget]: """All failed targets.""" return { target @@ -142,7 +169,7 @@ def failed_targets(self) -> set[Target]: } @property - def processing_targets(self) -> set[Target]: + def processing_targets(self) -> set[ImageTarget | VuMarkTarget]: """All processing targets.""" return { target diff --git a/src/mock_vws/database_type.py b/src/mock_vws/database_type.py new file mode 100644 index 000000000..a491eb62f --- /dev/null +++ b/src/mock_vws/database_type.py @@ -0,0 +1,14 @@ +"""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() + VUMARK = auto() diff --git a/src/mock_vws/target.py b/src/mock_vws/target.py index afd8f20b1..ac504b950 100644 --- a/src/mock_vws/target.py +++ b/src/mock_vws/target.py @@ -19,8 +19,8 @@ ) -class TargetDict(TypedDict): - """A dictionary type which represents a target.""" +class ImageTargetDict(TypedDict): + """A dictionary type which represents an image target.""" name: str width: float @@ -33,6 +33,20 @@ class TargetDict(TypedDict): delete_date_optional: str | None upload_date: str tracking_rating: int + target_type_name: str + + +class VuMarkTargetDict(TypedDict): + """A dictionary type which represents a VuMark template target.""" + + name: str + active_flag: bool + processing_time_seconds: float + target_id: str + last_modified_date: str + delete_date_optional: str | None + upload_date: str + target_type_name: str @beartype @@ -50,9 +64,8 @@ def _time_now() -> datetime.datetime: @beartype(conf=BeartypeConf(is_pep484_tower=True)) @dataclass(frozen=True, eq=True) -class Target: - """ - A Vuforia Target as managed in +class ImageTarget: + """A Vuforia image target as managed in https://developer.vuforia.com/target-manager. """ @@ -145,7 +158,7 @@ def tracking_rating(self) -> int: return self._post_processing_target_rating @classmethod - def from_dict(cls, target_dict: TargetDict) -> Self: + def from_dict(cls, target_dict: ImageTargetDict) -> Self: """Load a target from a dictionary.""" timezone = ZoneInfo(key="GMT") name = target_dict["name"] @@ -187,7 +200,7 @@ def from_dict(cls, target_dict: TargetDict) -> Self: target_tracking_rater=target_tracking_rater, ) - def to_dict(self) -> TargetDict: + def to_dict(self) -> ImageTargetDict: """Dump a target to a dictionary which can be loaded as JSON.""" delete_date: str | None = None if self.delete_date: @@ -207,4 +220,88 @@ def to_dict(self) -> TargetDict: "delete_date_optional": delete_date, "upload_date": self.upload_date.isoformat(), "tracking_rating": self.tracking_rating, + "target_type_name": "IMAGE", + } + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +@dataclass(frozen=True, eq=True) +class VuMarkTarget: + """A VuMark template target as managed in + https://developer.vuforia.com/target-manager. + """ + + name: str + active_flag: bool + processing_time_seconds: float + delete_date: datetime.datetime | None = None + last_modified_date: datetime.datetime = field(default_factory=_time_now) + target_id: str = field(default_factory=_random_hex) + upload_date: datetime.datetime = field(default_factory=_time_now) + + @property + def status(self) -> str: + """Return the status of the VuMark template target. + + VuMark templates always succeed after the processing time. + """ + processing_time = datetime.timedelta( + seconds=float(self.processing_time_seconds), + ) + + timezone = self.upload_date.tzinfo + now = datetime.datetime.now(tz=timezone) + time_since_change = now - self.last_modified_date + + if time_since_change <= processing_time: + return TargetStatuses.PROCESSING.value + + return TargetStatuses.SUCCESS.value + + @classmethod + def from_dict(cls, target_dict: VuMarkTargetDict) -> Self: + """Load a VuMark target from a dictionary.""" + timezone = ZoneInfo(key="GMT") + delete_date_optional = target_dict["delete_date_optional"] + if delete_date_optional is None: + delete_date = None + else: + delete_date = datetime.datetime.fromisoformat( + delete_date_optional, + ).replace(tzinfo=timezone) + + last_modified_date = datetime.datetime.fromisoformat( + target_dict["last_modified_date"], + ).replace(tzinfo=timezone) + upload_date = datetime.datetime.fromisoformat( + target_dict["upload_date"], + ).replace(tzinfo=timezone) + + return cls( + name=target_dict["name"], + active_flag=target_dict["active_flag"], + target_id=target_dict["target_id"], + processing_time_seconds=target_dict["processing_time_seconds"], + delete_date=delete_date, + last_modified_date=last_modified_date, + upload_date=upload_date, + ) + + def to_dict(self) -> VuMarkTargetDict: + """Dump a VuMark target to a dictionary which can be loaded as + JSON. + """ + delete_date: str | None = None + if self.delete_date: + delete_date = self.delete_date.isoformat() + + return { + "name": self.name, + "active_flag": self.active_flag, + "target_id": self.target_id, + "processing_time_seconds": float(self.processing_time_seconds), + "delete_date_optional": delete_date, + "last_modified_date": self.last_modified_date.isoformat(), + "upload_date": self.upload_date.isoformat(), + "target_type_name": "VUMARK_TEMPLATE", } diff --git a/tests/mock_vws/fixtures/vuforia_backends.py b/tests/mock_vws/fixtures/vuforia_backends.py index c0e95b08c..60c2f7bc3 100644 --- a/tests/mock_vws/fixtures/vuforia_backends.py +++ b/tests/mock_vws/fixtures/vuforia_backends.py @@ -20,11 +20,10 @@ from mock_vws._flask_server.vwq import CLOUDRECO_FLASK_APP from mock_vws._flask_server.vws import VWS_FLASK_APP from mock_vws.database import VuforiaDatabase +from mock_vws.database_type import DatabaseType from mock_vws.states import States -from mock_vws.target import Target -from mock_vws.target_raters import HardcodedTargetTrackingRater +from mock_vws.target import VuMarkTarget from tests.mock_vws.fixtures.credentials import VuMarkVuforiaDatabase -from tests.mock_vws.utils import make_image_file from tests.mock_vws.utils.retries import RETRY_ON_TOO_MANY_REQUESTS LOGGER = logging.getLogger(name=__name__) @@ -67,19 +66,10 @@ def _vumark_database( vumark_vuforia_database: VuMarkVuforiaDatabase, ) -> VuforiaDatabase: """Return a database with a target for VuMark instance generation.""" - vumark_target = Target( + vumark_target = VuMarkTarget( active_flag=True, - application_metadata=None, - image_value=make_image_file( - file_format="PNG", - color_space="RGB", - width=8, - height=8, - ).getvalue(), name="mock-vumark-target", processing_time_seconds=0, - width=1, - target_tracking_rater=HardcodedTargetTrackingRater(rating=5), target_id=vumark_vuforia_database.target_id, ) return VuforiaDatabase( @@ -87,6 +77,7 @@ def _vumark_database( server_access_key=vumark_vuforia_database.server_access_key, server_secret_key=vumark_vuforia_database.server_secret_key, targets={vumark_target}, + database_type=DatabaseType.VUMARK, ) diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index 4b9cf1f5c..211434649 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -18,7 +18,7 @@ from mock_vws import MissingSchemeError, MockVWS from mock_vws.database import VuforiaDatabase from mock_vws.image_matchers import ExactMatcher, StructuralSimilarityMatcher -from mock_vws.target import Target +from mock_vws.target import ImageTarget from tests.mock_vws.utils import Endpoint from tests.mock_vws.utils.usage_test_helpers import ( processing_time_seconds, @@ -395,12 +395,13 @@ 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 assert json.dumps(obj=target_dict) - new_target = Target.from_dict(target_dict=target_dict) + new_target = ImageTarget.from_dict(target_dict=target_dict) assert new_target == target @staticmethod @@ -431,12 +432,13 @@ 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 assert json.dumps(obj=target_dict) - new_target = Target.from_dict(target_dict=target_dict) + new_target = ImageTarget.from_dict(target_dict=target_dict) assert new_target.delete_date == target.delete_date diff --git a/tests/mock_vws/test_target_validators.py b/tests/mock_vws/test_target_validators.py index 04b147422..4e49ce89c 100644 --- a/tests/mock_vws/test_target_validators.py +++ b/tests/mock_vws/test_target_validators.py @@ -10,14 +10,14 @@ validate_target_id_exists, ) from mock_vws.database import VuforiaDatabase -from mock_vws.target import Target +from mock_vws.target import ImageTarget from mock_vws.target_raters import HardcodedTargetTrackingRater from tests.mock_vws.utils import make_image_file def _database_with_target(*, target_id: str) -> VuforiaDatabase: """Create a database containing one target with the given ID.""" - target = Target( + target = ImageTarget( active_flag=True, application_metadata=None, image_value=make_image_file( diff --git a/tests/mock_vws/test_vumark_generation_api.py b/tests/mock_vws/test_vumark_generation_api.py index 2b589ebd6..48334ad4a 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 VuforiaDatabase from tests.mock_vws.fixtures.credentials import VuMarkVuforiaDatabase +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: VuMarkVuforiaDatabase, + 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,40 @@ def test_empty_instance_id( response_json["result_code"] == ResultCodes.INVALID_INSTANCE_ID.value ) + + @staticmethod + def test_non_vumark_database( + vuforia_database: VuforiaDatabase, + ) -> 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.FORBIDDEN + response_json = response.json() + assert ( + response_json["result_code"] == ResultCodes.INVALID_TARGET_TYPE.value + )