From fdd66aea92eabdce9032faae65f7126aaa3f7c1a Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 03:00:06 +0000 Subject: [PATCH 1/3] Validate target status for VuMark instance generation (#2981) Return TargetStatusNotSuccessError when a VuMark instance generation request targets a VuMark target that is not yet in the success state. Co-authored-by: Cursor --- src/mock_vws/_flask_server/vws.py | 7 ++-- .../mock_web_services_api.py | 5 +++ src/mock_vws/database.py | 9 +++++ src/mock_vws/target.py | 38 +++++++++++++++++++ 4 files changed, 56 insertions(+), 3 deletions(-) diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index ab1c1cb9b..57942cb20 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -398,9 +398,6 @@ def generate_vumark_instance(target_id: str) -> Response: databases=all_databases, ) - # ``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, @@ -411,6 +408,10 @@ def generate_vumark_instance(target_id: str) -> Response: if not isinstance(database, VuMarkDatabase): raise InvalidTargetTypeError + target = database.get_vumark_target(target_id=target_id) + if target.status != TargetStatuses.SUCCESS.value: + raise TargetStatusNotSuccessError + accept = request.headers.get(key="Accept", default="") valid_accept_types: dict[str, bytes] = { "image/png": VUMARK_PNG, 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 671539cdb..50fc7aa95 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 @@ -319,6 +319,11 @@ def generate_vumark_instance(self, request: RequestData) -> _ResponseType: if not isinstance(database, VuMarkDatabase): raise InvalidTargetTypeError + target_id = request.path.split(sep="/")[-2] + target = database.get_vumark_target(target_id=target_id) + if target.status != TargetStatuses.SUCCESS.value: + raise TargetStatusNotSuccessError + accept = dict(request.headers).get("Accept", "") if accept not in valid_accept_types: raise InvalidAcceptHeaderError diff --git a/src/mock_vws/database.py b/src/mock_vws/database.py index d4e2389a8..61ec48377 100644 --- a/src/mock_vws/database.py +++ b/src/mock_vws/database.py @@ -203,6 +203,15 @@ class VuMarkDatabase: hash=False, ) + def get_vumark_target(self, target_id: str) -> VuMarkTarget: + """Return a VuMark target from the database with the given ID.""" + (target,) = ( + target + for target in self.vumark_targets + if target.target_id == target_id + ) + return target + def to_dict(self) -> VuMarkDatabaseDict: """Dump a VuMark database to a dictionary which can be loaded as JSON. diff --git a/src/mock_vws/target.py b/src/mock_vws/target.py index 50e9ec034..2e3760e4b 100644 --- a/src/mock_vws/target.py +++ b/src/mock_vws/target.py @@ -24,6 +24,9 @@ class VuMarkTargetDict(TypedDict): target_id: str name: str + processing_time_seconds: float + last_modified_date: str + upload_date: str class ImageTargetDict(TypedDict): @@ -228,14 +231,46 @@ class VuMarkTarget: """ name: str + processing_time_seconds: float = 0.0 target_id: str = field(default_factory=_random_hex) + last_modified_date: datetime.datetime = field(default_factory=_time_now) + upload_date: datetime.datetime = field(default_factory=_time_now) + + @property + def status(self) -> str: + """Return the status of the target. + + VuMark targets always succeed after processing. + """ + 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") + 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( target_id=target_dict["target_id"], name=target_dict["name"], + processing_time_seconds=target_dict["processing_time_seconds"], + last_modified_date=last_modified_date, + upload_date=upload_date, ) def to_dict(self) -> VuMarkTargetDict: @@ -245,4 +280,7 @@ def to_dict(self) -> VuMarkTargetDict: return { "target_id": self.target_id, "name": self.name, + "processing_time_seconds": float(self.processing_time_seconds), + "last_modified_date": self.last_modified_date.isoformat(), + "upload_date": self.upload_date.isoformat(), } From d0e068e95e80263045592e85e00ee4043dc520a8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 03:40:35 +0000 Subject: [PATCH 2/3] Add tests for VuMark target status validation and serialization Test that generating a VuMark instance for a still-processing target returns TargetStatusNotSuccessError, and that VuMarkTarget/VuMarkDatabase round-trip through to_dict/from_dict with the new timing fields. Co-authored-by: Cursor --- tests/mock_vws/test_requests_mock_usage.py | 37 ++++++- tests/mock_vws/test_vumark_generation_api.py | 101 ++++++++++++++++++- 2 files changed, 136 insertions(+), 2 deletions(-) diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index 87820d6ef..50d714e90 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 CloudDatabase, VuMarkDatabase from mock_vws.image_matchers import ExactMatcher, StructuralSimilarityMatcher -from mock_vws.target import ImageTarget +from mock_vws.target import ImageTarget, VuMarkTarget from tests.mock_vws.utils import Endpoint from tests.mock_vws.utils.usage_test_helpers import ( processing_time_seconds, @@ -441,6 +441,22 @@ def test_to_dict_deleted(high_quality_image: io.BytesIO) -> None: new_target = ImageTarget.from_dict(target_dict=target_dict) assert new_target.delete_date == target.delete_date + @staticmethod + def test_vumark_target_to_dict() -> None: + """It is possible to dump a VuMark target to a dictionary and + load it back. + """ + vumark_target = VuMarkTarget( + name="example-vumark", + processing_time_seconds=5.0, + ) + target_dict = vumark_target.to_dict() + + assert json.dumps(obj=target_dict) + + new_target = VuMarkTarget.from_dict(target_dict=target_dict) + assert new_target == vumark_target + class TestDatabaseToDict: """Tests for dumping a database to a dictionary.""" @@ -475,6 +491,25 @@ def test_to_dict(high_quality_image: io.BytesIO) -> None: new_database = CloudDatabase.from_dict(database_dict=database_dict) assert new_database == database + @staticmethod + def test_vumark_database_to_dict() -> None: + """It is possible to dump a VuMark database to a dictionary and + load it back. + """ + vumark_target = VuMarkTarget( + name="example-vumark", + processing_time_seconds=3.0, + ) + database = VuMarkDatabase( + vumark_targets={vumark_target}, + ) + + database_dict = database.to_dict() + assert json.dumps(obj=database_dict) + + new_database = VuMarkDatabase.from_dict(database_dict=database_dict) + assert new_database == database + class TestDateHeader: """Tests for the date header in responses from mock routes.""" diff --git a/tests/mock_vws/test_vumark_generation_api.py b/tests/mock_vws/test_vumark_generation_api.py index d78ab76d1..22e11b6a3 100644 --- a/tests/mock_vws/test_vumark_generation_api.py +++ b/tests/mock_vws/test_vumark_generation_api.py @@ -11,12 +11,15 @@ from vws.exceptions.vws_exceptions import ( InvalidInstanceIdError, InvalidTargetTypeError, + TargetStatusNotSuccessError, ) from vws.vumark_accept import VuMarkAccept from vws_auth_tools import authorization_header, rfc_1123_date +from mock_vws import MockVWS from mock_vws._constants import ResultCodes -from mock_vws.database import CloudDatabase +from mock_vws.database import CloudDatabase, VuMarkDatabase +from mock_vws.target import VuMarkTarget from tests.mock_vws.fixtures.credentials import VuMarkCloudDatabase from tests.mock_vws.utils import make_image_file @@ -259,3 +262,99 @@ def test_non_vumark_database( response_json["result_code"] == ResultCodes.INVALID_TARGET_TYPE.value ) + + +class TestTargetStatusNotSuccess: + """Tests for VuMark generation when the target is not in success + state. + """ + + @staticmethod + def test_processing_target() -> None: + """A VuMark target still processing returns + TargetStatusNotSuccess. + """ + vumark_target = VuMarkTarget( + name="processing-target", + processing_time_seconds=9999, + ) + vumark_database = VuMarkDatabase( + vumark_targets={vumark_target}, + ) + vumark_client = _make_vumark_service( + server_access_key=vumark_database.server_access_key, + server_secret_key=vumark_database.server_secret_key, + ) + + with MockVWS() as mock: + mock.add_vumark_database(vumark_database=vumark_database) + with pytest.raises( + expected_exception=TargetStatusNotSuccessError, + ) as exc: + vumark_client.generate_vumark_instance( + target_id=vumark_target.target_id, + instance_id=uuid4().hex, + accept=VuMarkAccept.PNG, + ) + + assert exc.value.response.status_code == HTTPStatus.FORBIDDEN + response_json = json.loads(s=exc.value.response.text) + assert ( + response_json["result_code"] + == ResultCodes.TARGET_STATUS_NOT_SUCCESS.value + ) + + @staticmethod + def test_processing_target_raw_response() -> None: + """The raw HTTP response for a processing target has the expected + status code and result code. + """ + vumark_target = VuMarkTarget( + name="processing-target", + processing_time_seconds=9999, + ) + vumark_database = VuMarkDatabase( + vumark_targets={vumark_target}, + ) + + with MockVWS() as mock: + mock.add_vumark_database(vumark_database=vumark_database) + response = _make_vumark_request( + server_access_key=vumark_database.server_access_key, + server_secret_key=vumark_database.server_secret_key, + target_id=vumark_target.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.TARGET_STATUS_NOT_SUCCESS.value + ) + + @staticmethod + def test_successful_target() -> None: + """A VuMark target that has finished processing succeeds.""" + vumark_target = VuMarkTarget( + name="ready-target", + processing_time_seconds=0, + ) + vumark_database = VuMarkDatabase( + vumark_targets={vumark_target}, + ) + vumark_client = _make_vumark_service( + server_access_key=vumark_database.server_access_key, + server_secret_key=vumark_database.server_secret_key, + ) + + with MockVWS() as mock: + mock.add_vumark_database(vumark_database=vumark_database) + vumark_bytes = vumark_client.generate_vumark_instance( + target_id=vumark_target.target_id, + instance_id=uuid4().hex, + accept=VuMarkAccept.PNG, + ) + + assert vumark_bytes.strip().startswith(_PNG_SIGNATURE) From 62501fa80428c6e966895ff3262aac6501032a49 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 09:47:43 +0000 Subject: [PATCH 3/3] Add Flask app test for VuMark target status validation TestVuMarkTargetStatus.test_processing_target_returns_forbidden exercises the Flask server code path (vws.py generate_vumark_instance) with a processing VuMark target, covering the branch that raises TargetStatusNotSuccessError. MockVWS uses the requests-mock server, so this branch was previously uncovered. Co-authored-by: Cursor --- tests/mock_vws/test_flask_app_usage.py | 82 +++++++++++++++++++++++++- 1 file changed, 81 insertions(+), 1 deletion(-) diff --git a/tests/mock_vws/test_flask_app_usage.py b/tests/mock_vws/test_flask_app_usage.py index bef5b44ec..6b7f7eee2 100644 --- a/tests/mock_vws/test_flask_app_usage.py +++ b/tests/mock_vws/test_flask_app_usage.py @@ -6,7 +6,7 @@ import time import uuid from collections.abc import Iterator -from http import HTTPStatus +from http import HTTPMethod, HTTPStatus import pytest import requests @@ -14,7 +14,9 @@ from PIL import Image from requests_mock_flask import add_flask_app_to_mock from vws import VWS, CloudRecoService +from vws_auth_tools import authorization_header, rfc_1123_date +from mock_vws._constants import ResultCodes from mock_vws._flask_server.target_manager import ( TARGET_MANAGER, TARGET_MANAGER_FLASK_APP, @@ -22,6 +24,7 @@ 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 CloudDatabase, VuMarkDatabase +from mock_vws.target import VuMarkTarget from tests.mock_vws.utils.usage_test_helpers import ( processing_time_seconds, ) @@ -709,6 +712,83 @@ def test_random( assert lowest_rating != highest_rating +class TestVuMarkTargetStatus: + """Tests for VuMark instance generation when target status is + validated (Flask app code path). + """ + + @staticmethod + def test_processing_target_returns_forbidden() -> None: + """A VuMark target still processing returns 403 when generating + an instance via the Flask app. + """ + vumark_target = VuMarkTarget( + name="processing-target", + processing_time_seconds=9999, + ) + vumark_database = VuMarkDatabase( + vumark_targets=set(), + ) + + vumark_databases_url = ( + _EXAMPLE_URL_FOR_TARGET_MANAGER + "/vumark_databases" + ) + response = requests.post( + url=vumark_databases_url, + json=vumark_database.to_dict(), + timeout=30, + ) + assert response.status_code == HTTPStatus.CREATED + database_data = json.loads(s=response.text) + + vumark_targets_url = ( + f"{vumark_databases_url}" + f"/{database_data['database_name']}/vumark_targets" + ) + response = requests.post( + url=vumark_targets_url, + json=vumark_target.to_dict(), + timeout=30, + ) + assert response.status_code == HTTPStatus.CREATED + + request_path = f"/targets/{vumark_target.target_id}/instances" + content_type = "application/json" + content = json.dumps( + obj={"instance_id": uuid.uuid4().hex}, + ).encode(encoding="utf-8") + date = rfc_1123_date() + authorization_string = authorization_header( + access_key=vumark_database.server_access_key, + secret_key=vumark_database.server_secret_key, + method=HTTPMethod.POST, + content=content, + content_type=content_type, + date=date, + request_path=request_path, + ) + + response = requests.post( + url="https://vws.vuforia.com" + 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.FORBIDDEN + response_json = response.json() + assert ( + response_json["result_code"] + == ResultCodes.TARGET_STATUS_NOT_SUCCESS.value + ) + + class TestResponseDelay: """Tests for the response delay feature.