Remove part done work

Author: adamtheturtle

src/vws/__init__.py

@@ -3,9 +3,11 @@
 """
 
 from ._version import get_versions
+from .query import CloudRecoService
 from .vws import VWS
 
 __all__ = [
+    'CloudRecoService',
     'VWS',
 ]
 

src/vws/query.py

@@ -0,0 +1,66 @@
+import io
+from urllib.parse import urljoin
+
+import requests
+from urllib3.filepost import encode_multipart_formdata
+
+from ._authorization import authorization_header, rfc_1123_date
+
+
+class CloudRecoService:
+    """
+    TODO
+    """
+
+    def __init__(
+        self,
+        client_access_key: str,
+        client_secret_key: str,
+    ) -> None:
+        """
+        Args:
+            client_access_key: A VWS client access key.
+            client_secret_key: A VWS client secret key.
+        """
+        self._client_access_key = client_access_key.encode()
+        self._client_secret_key = client_secret_key.encode()
+
+    def query(self, image: io.BytesIO) -> str:
+        """
+        TODO docstring
+        """
+        image_content = image.getvalue()
+        body = {
+            'image': ('image.jpeg', image_content, 'image/jpeg'),
+        }
+        date = rfc_1123_date()
+        request_path = '/v1/query'
+        content, content_type_header = encode_multipart_formdata(body)
+        method = 'POST'
+
+        authorization_string = authorization_header(
+            access_key=self._client_access_key,
+            secret_key=self._client_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,
+        }
+
+        base_vwq_url = 'https://cloudreco.vuforia.com'
+        response = requests.request(
+            method=method,
+            url=urljoin(base=base_vwq_url, url=request_path),
+            headers=headers,
+            data=content,
+        )
+
+        return response.json()['results']

tests/conftest.py

@@ -8,7 +8,7 @@
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
 
-from vws import VWS
+from vws import VWS, CloudRecoService
 
 pytest_plugins = [  # pylint: disable=invalid-name
     'tests.fixtures.images',
@@ -35,3 +35,14 @@ def vws_client(_mock_database: VuforiaDatabase) -> Iterator[VWS]:
         server_access_key=_mock_database.server_access_key,
         server_secret_key=_mock_database.server_secret_key,
     )
+
+
+@pytest.fixture()
+def cloud_reco_client(_mock_database: VuforiaDatabase) -> Iterator[VWS]:
+    """
+    TODO Docstring
+    """
+    yield CloudRecoService(
+        client_access_key=_mock_database.client_access_key,
+        client_secret_key=_mock_database.client_secret_key,
+    )

tests/test_query.py

@@ -0,0 +1,75 @@
+"""
+TODO
+"""
+
+import io
+import uuid
+
+import pytest
+
+from vws import VWS, CloudRecoService
+
+
+class TestQuery:
+    """
+    TODO
+    """
+
+    def test_no_matches(
+        self,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        An empty list is returned if there are no matches.
+        """
+        result = cloud_reco_client.query(image=high_quality_image)
+        assert result == []
+
+    def test_match(
+        self,
+        vws_client: VWS,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        Details of matching targets are returned.
+        """
+        target_id = vws_client.add_target(
+            name='x',
+            width=1,
+            image=high_quality_image,
+        )
+        vws_client.wait_for_target_processed(target_id=target_id)
+        [matching_target] = cloud_reco_client.query(image=high_quality_image)
+        assert matching_target['target_id'] == target_id
+
+
+class TestMaxNumResults:
+    """
+    Tests for the ``max_num_results`` parameter of ``query``.
+    """
+
+    def test_default(
+        self,
+        vws_client: VWS,
+        cloud_reco_client: CloudRecoService,
+        high_quality_image: io.BytesIO,
+    ) -> None:
+        """
+        By default the maximum number of results is 1.
+        """
+        target_id = vws_client.add_target(
+            name=uuid.uuid4().hex,
+            width=1,
+            image=high_quality_image,
+        )
+        target_id_2 = vws_client.add_target(
+            name=uuid.uuid4().hex,
+            width=1,
+            image=high_quality_image,
+        )
+        vws_client.wait_for_target_processed(target_id=target_id)
+        vws_client.wait_for_target_processed(target_id=target_id_2)
+        matches = cloud_reco_client.query(image=high_quality_image)
+        assert len(matches) == 1