Merge pull request #630 from adamtheturtle/user-set-time

Allow users to set the amount of time after a target is deleted for which a 500 will be returned on a match

Author: adamtheturtle

README.rst

@@ -182,7 +182,8 @@ Matching deleted targets
 
 Matching a target which has been deleted returns a 500 (INTERNAL SERVER ERROR) response within the first few seconds.
 This timeframe is not consistent on the real Vuforia Web Services.
-On the mock, this timeframe is always three seconds.
+On the mock, this timeframe is three seconds by default.
+``MockVWS`` takes a parameter ``query_recognizes_deletion_seconds`` to change this.
 
 .. |Build Status| image:: https://travis-ci.org/adamtheturtle/vws-python.svg?branch=master
    :target: https://travis-ci.org/adamtheturtle/vws-python

src/mock_vws/__init__.py

@@ -37,6 +37,7 @@ def __init__(  # pylint: disable=too-many-arguments
         client_secret_key: Optional[str]=None,
         database_name: Optional[str]=None,
         processing_time_seconds: Union[int, float]=0.5,
+        query_recognizes_deletion_seconds: Union[int, float]=3,
     ) -> None:
         """
         Args:
@@ -56,6 +57,9 @@ def __init__(  # pylint: disable=too-many-arguments
                 deterministic.
             base_vwq_url: The base URL for the VWQ API.
             base_vws_url: The base URL for the VWS API.
+            query_recognizes_deletion_seconds: The number of seconds after a
+                target has been deleted that the query endpoint will return a
+                500 response for on a match.
 
 
         Attributes:
@@ -95,6 +99,10 @@ def __init__(  # pylint: disable=too-many-arguments
         self._base_vws_url = base_vws_url
         self._base_vwq_url = base_vwq_url
 
+        self._query_recognizes_deletion_seconds = (
+            query_recognizes_deletion_seconds
+        )
+
     def __enter__(self) -> 'MockVWS':
         """
         Start an instance of a Vuforia mock with access keys set from
@@ -115,7 +123,9 @@ def __enter__(self) -> 'MockVWS':
             client_access_key=self.client_access_key,
             client_secret_key=self.client_secret_key,
             mock_web_services_api=mock_vws_api,
-            query_recognizes_deletion_seconds=3,
+            query_recognizes_deletion_seconds=(
+                self._query_recognizes_deletion_seconds
+            ),
         )
 
         date = email.utils.formatdate(None, localtime=False, usegmt=True)

src/mock_vws/_mock_web_query_api.py

@@ -10,7 +10,7 @@
 import io
 import uuid
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Set, Tuple
+from typing import Any, Callable, Dict, List, Set, Tuple, Union
 
 import pytz
 import wrapt
@@ -449,7 +449,7 @@ def __init__(
         client_access_key: str,
         client_secret_key: str,
         mock_web_services_api: MockVuforiaWebServicesAPI,
-        query_recognizes_deletion_seconds: int,
+        query_recognizes_deletion_seconds: Union[int, float],
     ) -> None:
         """
         Args:

tests/mock_vws/test_query.py

@@ -8,12 +8,12 @@
 import calendar
 import io
 import time
-from typing import Any, Dict, Union
+from typing import Dict, Union
 from urllib.parse import urljoin
 
 import pytest
 import requests
-from requests import Response, codes
+from requests import codes
 from requests_mock import POST
 from urllib3.filepost import encode_multipart_formdata
 
@@ -22,6 +22,7 @@
     add_target_to_vws,
     delete_target,
     get_vws_target,
+    query,
     update_target,
     wait_for_target_processed,
 )
@@ -38,55 +39,6 @@
 VWQ_HOST = 'https://cloudreco.vuforia.com'
 
 
-def query(
-    vuforia_database_keys: VuforiaDatabaseKeys,
-    body: Dict[str, Any],
-) -> Response:
-    """
-    Make a request to the endpoint to make an image recognition query.
-
-    Args:
-        vuforia_database_keys: The credentials to use to connect to
-            Vuforia.
-        body: The request body to send in ``multipart/formdata`` format.
-
-    Returns:
-        The response returned by the API.
-    """
-    date = rfc_1123_date()
-    request_path = '/v1/query'
-    content, content_type_header = encode_multipart_formdata(body)
-    method = POST
-
-    access_key = vuforia_database_keys.client_access_key
-    secret_key = vuforia_database_keys.client_secret_key
-    authorization_string = authorization_header(
-        access_key=access_key,
-        secret_key=secret_key,
-        method=method,
-        content=content,
-        # Note that this is not the actual Content-Type header value sent.
-        content_type='multipart/form-data',
-        date=date,
-        request_path=request_path,
-    )
-
-    headers = {
-        'Authorization': authorization_string,
-        'Date': date,
-        'Content-Type': content_type_header,
-    }
-
-    response = requests.request(
-        method=method,
-        url=urljoin(base=VWQ_HOST, url=request_path),
-        headers=headers,
-        data=content,
-    )
-
-    return response
-
-
 @pytest.mark.usefixtures('verify_mock_vuforia')
 class TestContentType:
     """

tests/mock_vws/test_usage.py

@@ -7,6 +7,7 @@
 import io
 import socket
 import string
+import time
 import uuid
 
 import pytest
@@ -22,9 +23,13 @@
     VuforiaDatabaseKeys,
     add_target_to_vws,
     database_summary,
+    delete_target,
     get_vws_target,
+    query,
     rfc_1123_date,
+    wait_for_target_processed,
 )
+from tests.mock_vws.utils.assertions import assert_query_success
 
 
 def request_unmocked_address() -> None:
@@ -420,3 +425,119 @@ def test_custom_base_vwq_url(self) -> None:
 
             requests.post(url='https://vuforia.vwq.example.com/v1/query')
             requests.get('https://vws.vuforia.com/summary')
+
+
+class TestCustomQueryRecognizesDeletionSeconds:
+    """
+    Tests for setting the amount of time after a target has been deleted
+    until it is not recognized by the query endpoint.
+    """
+
+    def _recognize_deletion_seconds(
+        self,
+        high_quality_image: io.BytesIO,
+        vuforia_database_keys: VuforiaDatabaseKeys,
+    ) -> float:
+        """
+        XXX
+        """
+        image_content = high_quality_image.getvalue()
+        image_data_encoded = base64.b64encode(image_content).decode('ascii')
+        add_target_data = {
+            'name': 'example_name',
+            'width': 1,
+            'image': image_data_encoded,
+        }
+        response = add_target_to_vws(
+            vuforia_database_keys=vuforia_database_keys,
+            data=add_target_data,
+        )
+
+        target_id = response.json()['target_id']
+
+        wait_for_target_processed(
+            target_id=target_id,
+            vuforia_database_keys=vuforia_database_keys,
+        )
+
+        response = delete_target(
+            vuforia_database_keys=vuforia_database_keys,
+            target_id=target_id,
+        )
+
+        time_after_delete = datetime.datetime.now()
+
+        body = {'image': ('image.jpeg', image_content, 'image/jpeg')}
+
+        while True:
+            response = query(
+                vuforia_database_keys=vuforia_database_keys,
+                body=body,
+            )
+
+            try:
+                assert_query_success(response=response)
+            except AssertionError:
+                # The response text for a 500 response is not consistent.
+                # Therefore we only test for consistent features.
+                assert 'Error 500 Server Error' in response.text
+                assert 'HTTP ERROR 500' in response.text
+                assert 'Problem accessing /v1/query' in response.text
+                time.sleep(0.05)
+                continue
+
+            assert response.json()['results'] == []
+            time_difference = datetime.datetime.now() - time_after_delete
+            return time_difference.total_seconds()
+
+    def test_default(
+        self,
+        high_quality_image: io.BytesIO,
+        vuforia_database_keys: VuforiaDatabaseKeys,
+    ) -> None:
+        """
+        By default it takes three seconds for the Query API on the mock to
+        recognize that a target has been deleted.
+
+        The real Query API takes between seven and thirty seconds.
+        See ``test_query`` for more information.
+        """
+        with MockVWS(
+            client_access_key=vuforia_database_keys.client_access_key.decode(),
+            client_secret_key=vuforia_database_keys.client_secret_key.decode(),
+            server_access_key=vuforia_database_keys.server_access_key.decode(),
+            server_secret_key=vuforia_database_keys.server_secret_key.decode(),
+        ):
+            recognize_deletion_seconds = self._recognize_deletion_seconds(
+                high_quality_image=high_quality_image,
+                vuforia_database_keys=vuforia_database_keys,
+            )
+
+        expected = 3
+        assert abs(expected - recognize_deletion_seconds) < 0.2
+
+    def test_custom(
+        self,
+        high_quality_image: io.BytesIO,
+        vuforia_database_keys: VuforiaDatabaseKeys,
+    ) -> None:
+        """
+        It is possible to use set a custom amount of time that it takes for the
+        Query API on the mock to recognize that a target has been deleted.
+        """
+        # We choose a low time for a quick test.
+        query_recognizes_deletion = 0.1
+        with MockVWS(
+            client_access_key=vuforia_database_keys.client_access_key.decode(),
+            client_secret_key=vuforia_database_keys.client_secret_key.decode(),
+            server_access_key=vuforia_database_keys.server_access_key.decode(),
+            server_secret_key=vuforia_database_keys.server_secret_key.decode(),
+            query_recognizes_deletion_seconds=query_recognizes_deletion,
+        ):
+            recognize_deletion_seconds = self._recognize_deletion_seconds(
+                high_quality_image=high_quality_image,
+                vuforia_database_keys=vuforia_database_keys,
+            )
+
+        expected = query_recognizes_deletion
+        assert abs(expected - recognize_deletion_seconds) < 0.2

tests/mock_vws/utils/__init__.py

@@ -11,6 +11,7 @@
 import timeout_decorator
 from requests import Response
 from requests_mock import DELETE, GET, POST, PUT
+from urllib3.filepost import encode_multipart_formdata
 
 from mock_vws._constants import ResultCodes, TargetStatuses
 from tests.mock_vws.utils.authorization import (
@@ -366,3 +367,53 @@ def target_summary(
     )
 
     return response
+
+
+def query(
+    vuforia_database_keys: VuforiaDatabaseKeys,
+    body: Dict[str, Any],
+) -> Response:
+    """
+    Make a request to the endpoint to make an image recognition query.
+
+    Args:
+        vuforia_database_keys: The credentials to use to connect to
+            Vuforia.
+        body: The request body to send in ``multipart/formdata`` format.
+
+    Returns:
+        The response returned by the API.
+    """
+    date = rfc_1123_date()
+    request_path = '/v1/query'
+    content, content_type_header = encode_multipart_formdata(body)
+    method = POST
+
+    access_key = vuforia_database_keys.client_access_key
+    secret_key = vuforia_database_keys.client_secret_key
+    authorization_string = authorization_header(
+        access_key=access_key,
+        secret_key=secret_key,
+        method=method,
+        content=content,
+        # Note that this is not the actual Content-Type header value sent.
+        content_type='multipart/form-data',
+        date=date,
+        request_path=request_path,
+    )
+
+    headers = {
+        'Authorization': authorization_string,
+        'Date': date,
+        'Content-Type': content_type_header,
+    }
+
+    vwq_host = 'https://cloudreco.vuforia.com'
+    response = requests.request(
+        method=method,
+        url=urljoin(base=vwq_host, url=request_path),
+        headers=headers,
+        data=content,
+    )
+
+    return response