Move query helper to utils

Author: adamtheturtle

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,8 +23,10 @@
     VuforiaDatabaseKeys,
     add_target_to_vws,
     database_summary,
+    delete_target,
     get_vws_target,
     rfc_1123_date,
+    wait_for_target_processed,
 )
 
 
@@ -428,5 +431,84 @@ class TestCustomQueryRecognizesDeletionSeconds:
     until it is not recognized by the query endpoint.
     """
 
-    def test_default(self):
-        pass
+    def _time_to_recognize_deletion(
+        self,
+        high_quality_image: io.BytesIO,
+        vuforia_database_keys: VuforiaDatabaseKeys,
+    ) -> datetime.timedelta:
+        """
+        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'] == []
+            return datetime.datetime.now() - time_after_delete
+
+    def test_default(
+        self,
+        high_quality_image: io.BytesIO,
+        vuforia_database_keys: VuforiaDatabaseKeys,
+    ) -> None:
+        """
+        XXX
+        """
+        with MockVWS(
+            base_vwq_url='https://vuforia.vwq.example.com',
+            real_http=False,
+            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(),
+        ) as mock:
+            time_to_recognize_deletion = self._time_to_recognize_deletion(
+                high_quality_image=high_quality_image,
+                vuforia_database_keys=vuforia_database_keys,
+            )
+
+        import pdb
+        pdb.set_trace()
+        expected == 3
+        assert abs(expected - 1) == 3

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,54 @@ 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