Merge branch 'more-file-types' into more-beartype

Author: adamtheturtle

README.rst

@@ -30,7 +30,8 @@ Getting Started
    from mock_vws import MockVWS
    from mock_vws.database import VuforiaDatabase
 
-   mock = MockVWS(real_http=False)
+   # We use a low processing time so that tests run quickly.
+   mock = MockVWS(processing_time_seconds=0.2)
    database = VuforiaDatabase(
        server_access_key='[server-access-key]',
        server_secret_key='[server-secret-key]',

src/vws/query.py

@@ -3,10 +3,10 @@
 """
 
 import datetime
+import io
 import json
 from http import HTTPMethod, HTTPStatus
-from io import BufferedRandom, BytesIO
-from typing import Any
+from typing import Any, BinaryIO
 from urllib.parse import urljoin
 
 import requests
@@ -29,9 +29,11 @@
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult, TargetData
 
+_ImageType = io.BytesIO | BinaryIO
+
 
 @beartype
-def _get_image_data(image: BytesIO | BufferedRandom) -> bytes:
+def _get_image_data(image: _ImageType) -> bytes:
     """Get the data of an image file."""
     original_tell = image.tell()
     image.seek(0)
@@ -64,7 +66,7 @@ def __init__(
 
     def query(
         self,
-        image: BytesIO | BufferedRandom,
+        image: _ImageType,
         max_num_results: int = 1,
         include_target_data: CloudRecoIncludeTargetData = (
             CloudRecoIncludeTargetData.TOP

src/vws/vws.py

@@ -8,7 +8,7 @@
 import time
 from datetime import date
 from http import HTTPMethod, HTTPStatus
-from io import BufferedRandom, BytesIO
+from typing import BinaryIO
 from urllib.parse import urljoin
 
 import requests
@@ -49,9 +49,11 @@
 
 from .exceptions.response import Response
 
+_ImageType = io.BytesIO | BinaryIO
+
 
 @beartype
-def _get_image_data(image: BytesIO | BufferedRandom) -> bytes:
+def _get_image_data(image: _ImageType) -> bytes:
     """Get the data of an image file."""
     original_tell = image.tell()
     image.seek(0)
@@ -238,7 +240,7 @@ def add_target(
         self,
         name: str,
         width: float,
-        image: BytesIO | BufferedRandom,
+        image: _ImageType,
         application_metadata: str | None,
         *,
         active_flag: bool,
@@ -621,7 +623,7 @@ def update_target(
         target_id: str,
         name: str | None = None,
         width: float | None = None,
-        image: io.BytesIO | io.BufferedRandom | None = None,
+        image: _ImageType | None = None,
         active_flag: bool | None = None,
         application_metadata: str | None = None,
     ) -> None:

tests/conftest.py

@@ -5,6 +5,7 @@
 import io
 from collections.abc import Generator
 from pathlib import Path
+from typing import BinaryIO, Literal
 
 import pytest
 from mock_vws import MockVWS
@@ -18,7 +19,8 @@ def mock_database() -> Generator[VuforiaDatabase, None, None]:
     """
     Yield a mock ``VuforiaDatabase``.
     """
-    with MockVWS() as mock:
+    # We use a low processing time so that tests run quickly.
+    with MockVWS(processing_time_seconds=0.2) as mock:
         database = VuforiaDatabase()
         mock.add_database(database=database)
         yield database
@@ -46,25 +48,28 @@ def cloud_reco_client(_mock_database: VuforiaDatabase) -> CloudRecoService:
     )
 
 
-@pytest.fixture
-def image_file(
+@pytest.fixture(name="image_file", params=["r+b", "rb"])
+def image_file_fixture(
     high_quality_image: io.BytesIO,
     tmp_path: Path,
-) -> Generator[io.BufferedRandom, None, None]:
+    request: pytest.FixtureRequest,
+) -> Generator[BinaryIO, None, None]:
     """An image file object."""
     file = tmp_path / "image.jpg"
-    file.touch()
-    with file.open("r+b") as fileobj:
-        buffer = high_quality_image.getvalue()
-        fileobj.write(buffer)
-        yield fileobj
+    buffer = high_quality_image.getvalue()
+    file.write_bytes(data=buffer)
+    mode: Literal["r+b", "rb"] = request.param
+    with file.open(mode=mode) as file_obj:
+        yield file_obj
 
 
 @pytest.fixture(params=["high_quality_image", "image_file"])
 def image(
     request: pytest.FixtureRequest,
-) -> io.BytesIO | io.BufferedRandom:
+    high_quality_image: io.BytesIO,
+    image_file: BinaryIO,
+) -> io.BytesIO | BinaryIO:
     """An image in any of the types that the API accepts."""
-    result = request.getfixturevalue(request.param)
-    assert isinstance(result, io.BytesIO | io.BufferedRandom)
-    return result
+    if request.param == "high_quality_image":
+        return high_quality_image
+    return image_file

tests/test_cloud_reco_exceptions.py

@@ -27,7 +27,7 @@
 
 def test_too_many_max_results(
     cloud_reco_client: CloudRecoService,
-    high_quality_image: io.BytesIO | io.BufferedRandom,
+    high_quality_image: io.BytesIO,
 ) -> None:
     """
     A ``MaxNumResultsOutOfRange`` error is raised if the given
@@ -78,7 +78,7 @@ def test_cloudrecoexception_inheritance() -> None:
 
 
 def test_authentication_failure(
-    high_quality_image: io.BytesIO | io.BufferedRandom,
+    high_quality_image: io.BytesIO,
 ) -> None:
     """
     An ``AuthenticationFailure`` exception is raised when the client access key
@@ -99,7 +99,7 @@ def test_authentication_failure(
 
 
 def test_inactive_project(
-    high_quality_image: io.BytesIO | io.BufferedRandom,
+    high_quality_image: io.BytesIO,
 ) -> None:
     """
     An ``InactiveProject`` exception is raised when querying an inactive

tests/test_query.py

@@ -4,6 +4,7 @@
 
 import io
 import uuid
+from typing import BinaryIO
 
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
@@ -20,7 +21,7 @@ class TestQuery:
     @staticmethod
     def test_no_matches(
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         An empty list is returned if there are no matches.
@@ -32,7 +33,7 @@ def test_no_matches(
     def test_match(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         Details of matching targets are returned.
@@ -55,7 +56,7 @@ class TestCustomBaseVWQURL:
     """
 
     @staticmethod
-    def test_custom_base_url(image: io.BytesIO | io.BufferedRandom) -> None:
+    def test_custom_base_url(image: io.BytesIO | BinaryIO) -> None:
         """
         It is possible to use query a target to a database under a custom VWQ
         URL.
@@ -100,7 +101,7 @@ class TestMaxNumResults:
     def test_default(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         By default the maximum number of results is 1.
@@ -128,7 +129,7 @@ def test_default(
     def test_custom(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to set a custom ``max_num_results``.
@@ -174,7 +175,7 @@ class TestIncludeTargetData:
     def test_default(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         By default, target data is only returned in the top match.
@@ -206,7 +207,7 @@ def test_default(
     def test_top(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.TOP`` is given, target data is only
@@ -240,7 +241,7 @@ def test_top(
     def test_none(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.NONE`` is given, target data is not
@@ -274,7 +275,7 @@ def test_none(
     def test_all(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.ALL`` is given, target data is

tests/test_vws.py

@@ -7,6 +7,7 @@
 import io
 import secrets
 import uuid
+from typing import BinaryIO
 
 import pytest
 from freezegun import freeze_time
@@ -33,7 +34,7 @@ class TestAddTarget:
     @pytest.mark.parametrize("active_flag", [True, False])
     def test_add_target(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
         application_metadata: bytes | None,
         cloud_reco_client: CloudRecoService,
         *,
@@ -77,7 +78,7 @@ def test_add_target(
     @staticmethod
     def test_add_two_targets(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         No exception is raised when adding two targets with different names.
@@ -100,7 +101,7 @@ class TestCustomBaseVWSURL:
     """
 
     @staticmethod
-    def test_custom_base_url(image: io.BytesIO | io.BufferedRandom) -> None:
+    def test_custom_base_url(image: io.BytesIO | BinaryIO) -> None:
         """
         It is possible to use add a target to a database under a custom VWS
         URL.
@@ -132,7 +133,7 @@ class TestListTargets:
     @staticmethod
     def test_list_targets(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to get a list of target IDs.
@@ -162,7 +163,7 @@ class TestDelete:
     @staticmethod
     def test_delete_target(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to delete a target.
@@ -189,7 +190,7 @@ class TestGetTargetSummaryReport:
     @staticmethod
     def test_get_target_summary_report(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         Details of a target are returned by ``get_target_summary_report``.
@@ -258,7 +259,7 @@ class TestGetTargetRecord:
     @staticmethod
     def test_get_target_record(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         Details of a target are returned by ``get_target_record``.
@@ -293,7 +294,7 @@ class TestWaitForTargetProcessed:
     @staticmethod
     def test_wait_for_target_processed(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to wait until a target is processed.
@@ -313,7 +314,7 @@ def test_wait_for_target_processed(
 
     @staticmethod
     def test_default_seconds_between_requests(
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         By default, 0.2 seconds are waited between polling requests.
@@ -365,7 +366,7 @@ def test_default_seconds_between_requests(
 
     @staticmethod
     def test_custom_seconds_between_requests(
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to customize the time waited between polling requests.
@@ -416,7 +417,7 @@ def test_custom_seconds_between_requests(
             assert report.request_usage == expected_requests
 
     @staticmethod
-    def test_custom_timeout(image: io.BytesIO | io.BufferedRandom) -> None:
+    def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None:
         """
         It is possible to set a maximum timeout.
         """
@@ -460,7 +461,7 @@ class TestGetDuplicateTargets:
     @staticmethod
     def test_get_duplicate_targets(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to get the IDs of similar targets.
@@ -494,8 +495,8 @@ class TestUpdateTarget:
     @staticmethod
     def test_update_target(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
-        different_high_quality_image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
+        different_high_quality_image: io.BytesIO,
         cloud_reco_client: CloudRecoService,
     ) -> None:
         """
@@ -553,7 +554,7 @@ def test_update_target(
     @staticmethod
     def test_no_fields_given(
         vws_client: VWS,
-        image: io.BytesIO | io.BufferedRandom,
+        image: io.BytesIO | BinaryIO,
     ) -> None:
         """
         It is possible to give no update fields.