Merge pull request #2554 from VWS-Python/fix-tests

Corrupted images now get an error

Author: adamtheturtle

src/mock_vws/_services_validators/__init__.py

@@ -29,6 +29,7 @@
     validate_image_data_type,
     validate_image_encoding,
     validate_image_format,
+    validate_image_integrity,
     validate_image_is_image,
     validate_image_size,
 )
@@ -122,6 +123,7 @@ def run_services_validators(
     validate_image_format(request_body=request_body)
     validate_image_color_space(request_body=request_body)
     validate_image_size(request_body=request_body)
+    validate_image_integrity(request_body=request_body)
 
     validate_name_type(request_body=request_body)
     validate_name_length(request_body=request_body)

src/mock_vws/_services_validators/image_validators.py

@@ -21,6 +21,36 @@
 _LOGGER = logging.getLogger(name=__name__)
 
 
+@beartype
+def validate_image_integrity(*, request_body: bytes) -> None:
+    """Validate the integrity of the image given to a VWS endpoint.
+
+    Args:
+        request_body: The body of the request.
+
+    Raises:
+        BadImageError: The image is given and is not a valid image file.
+    """
+    if not request_body:
+        return
+
+    request_text = request_body.decode()
+    image = json.loads(s=request_text).get("image")
+    if image is None:
+        return
+
+    decoded = decode_base64(encoded_data=image)
+
+    image_file = io.BytesIO(initial_bytes=decoded)
+    pil_image = Image.open(fp=image_file)
+
+    try:
+        pil_image.verify()
+    except SyntaxError as exc:
+        _LOGGER.warning(msg="The image is not a valid image file.")
+        raise BadImageError from exc
+
+
 @beartype
 def validate_image_format(*, request_body: bytes) -> None:
     """Validate the format of the image given to a VWS endpoint.

tests/mock_vws/test_add_target.py

@@ -478,14 +478,21 @@ def test_corrupted(
         vws_client: VWS,
     ) -> None:
         """
-        No error is returned when the given image is corrupted.
+        An error is returned when the given image is corrupted.
         """
-        vws_client.add_target(
-            name="example_name",
-            width=1,
-            image=corrupted_image_file,
-            application_metadata=None,
-            active_flag=True,
+        with pytest.raises(expected_exception=BadImageError) as exc:
+            vws_client.add_target(
+                name="example_name",
+                width=1,
+                image=corrupted_image_file,
+                application_metadata=None,
+                active_flag=True,
+            )
+
+        assert_vws_failure(
+            response=exc.value.response,
+            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+            result_code=ResultCodes.BAD_IMAGE,
         )
 
     @staticmethod

tests/mock_vws/test_flask_app_usage.py

@@ -455,7 +455,7 @@ class TestTargetRaters:
 
     @staticmethod
     def test_default(
-        corrupted_image_file: io.BytesIO,
+        image_file_success_state_low_rating: io.BytesIO,
         high_quality_image: io.BytesIO,
     ) -> None:
         """
@@ -470,10 +470,10 @@ def test_default(
             server_secret_key=database.server_secret_key,
         )
 
-        corrupted_image_target_id = vws_client.add_target(
+        low_rating_image_target_id = vws_client.add_target(
             name=uuid.uuid4().hex,
             width=1,
-            image=corrupted_image_file,
+            image=image_file_success_state_low_rating,
             application_metadata=None,
             active_flag=True,
         )
@@ -487,27 +487,26 @@ def test_default(
         )
 
         for target_id in (
-            corrupted_image_target_id,
+            low_rating_image_target_id,
             high_quality_image_target_id,
         ):
             vws_client.wait_for_target_processed(target_id=target_id)
 
-        corrupted_image_rating = vws_client.get_target_record(
-            target_id=corrupted_image_target_id,
+        low_rated_image_rating = vws_client.get_target_record(
+            target_id=low_rating_image_target_id,
         ).target_record.tracking_rating
 
         high_quality_image_rating = vws_client.get_target_record(
             target_id=high_quality_image_target_id,
         ).target_record.tracking_rating
 
-        # In the real Vuforia, this image may rate as -2.
-        assert corrupted_image_rating <= 0
+        assert low_rated_image_rating <= 0
         assert high_quality_image_rating > 1
 
     @staticmethod
     def test_brisque(
         monkeypatch: pytest.MonkeyPatch,
-        corrupted_image_file: io.BytesIO,
+        image_file_success_state_low_rating: io.BytesIO,
         high_quality_image: io.BytesIO,
     ) -> None:
         """
@@ -524,10 +523,10 @@ def test_brisque(
             server_secret_key=database.server_secret_key,
         )
 
-        corrupted_image_target_id = vws_client.add_target(
+        low_rating_image_target_id = vws_client.add_target(
             name=uuid.uuid4().hex,
             width=1,
-            image=corrupted_image_file,
+            image=image_file_success_state_low_rating,
             application_metadata=None,
             active_flag=True,
         )
@@ -541,21 +540,20 @@ def test_brisque(
         )
 
         for target_id in (
-            corrupted_image_target_id,
+            low_rating_image_target_id,
             high_quality_image_target_id,
         ):
             vws_client.wait_for_target_processed(target_id=target_id)
 
-        corrupted_image_rating = vws_client.get_target_record(
-            target_id=corrupted_image_target_id,
+        low_rated_image_rating = vws_client.get_target_record(
+            target_id=low_rating_image_target_id,
         ).target_record.tracking_rating
 
         high_quality_image_rating = vws_client.get_target_record(
             target_id=high_quality_image_target_id,
         ).target_record.tracking_rating
 
-        # In the real Vuforia, this image may rate as -2.
-        assert corrupted_image_rating <= 0
+        assert low_rated_image_rating <= 0
         assert high_quality_image_rating > 1
 
     @staticmethod

tests/mock_vws/test_get_target.py

@@ -146,7 +146,6 @@ def test_target_quality(
         vws_client: VWS,
         high_quality_image: io.BytesIO,
         image_file_success_state_low_rating: io.BytesIO,
-        corrupted_image_file: io.BytesIO,
     ) -> None:
         """
         The target tracking rating is as expected.
@@ -159,14 +158,9 @@ def test_target_quality(
             vws_client=vws_client,
             image_file=image_file_success_state_low_rating,
         )
-        corrupted_image_file_tracking_rating = _get_target_tracking_rating(
-            vws_client=vws_client,
-            image_file=corrupted_image_file,
-        )
         assert (
             high_quality_image_tracking_rating
             > low_quality_image_tracking_rating
-            >= corrupted_image_file_tracking_rating
         )
 
 

tests/mock_vws/test_target_raters.py

@@ -50,14 +50,15 @@ class TestBrisqueTargetTrackingRater:
     """
 
     @staticmethod
-    def test_low_quality_image(corrupted_image_file: io.BytesIO) -> None:
+    def test_low_quality_image(
+        image_file_success_state_low_rating: io.BytesIO,
+    ) -> None:
         """
         Test that a low quality image returns a low rating.
         """
         rater = BrisqueTargetTrackingRater()
-        image_content = corrupted_image_file.getvalue()
+        image_content = image_file_success_state_low_rating.getvalue()
         rating = rater(image_content=image_content)
-        # In the real Vuforia, this image may rate as -2.
         assert rating == 0
 
     @staticmethod

tests/mock_vws/test_update_target.py

@@ -670,12 +670,19 @@ def test_corrupted(
         target_id: str,
     ) -> None:
         """
-        No error is returned when the given image is corrupted.
+        An error is returned when the given image is corrupted.
         """
         vws_client.wait_for_target_processed(target_id=target_id)
-        vws_client.update_target(
-            target_id=target_id,
-            image=corrupted_image_file,
+        with pytest.raises(expected_exception=BadImageError) as exc:
+            vws_client.update_target(
+                target_id=target_id,
+                image=corrupted_image_file,
+            )
+
+        assert_vws_failure(
+            response=exc.value.response,
+            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+            result_code=ResultCodes.BAD_IMAGE,
         )
 
     @staticmethod