From ceba3428786c138953aee2f61ef9b0c79c19704c Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Mon, 12 Oct 2026 23:54:51 +0100 Subject: [PATCH] Replace request.param fixture selection with typed parameters --- conftest.py | 7 +- pyproject.toml | 7 +- tests/async_vws/test_configuration.py | 13 ++- tests/async_vws/test_processing.py | 23 +++- tests/async_vws/test_reports.py | 26 ++++- tests/async_vws/test_targets.py | 83 ++++++++++++-- tests/conftest.py | 64 +++-------- tests/image_factory.py | 55 ++++++++++ tests/test_async_query.py | 76 +++++++++++-- tests/test_query.py | 151 +++++++++++++++++++++++--- tests/vws/test_configuration.py | 35 +++++- tests/vws/test_processing.py | 45 +++++++- tests/vws/test_reports.py | 24 +++- tests/vws/test_targets.py | 83 ++++++++++++-- 14 files changed, 575 insertions(+), 117 deletions(-) create mode 100644 tests/image_factory.py diff --git a/conftest.py b/conftest.py index 7869bab15..51612fc9f 100644 --- a/conftest.py +++ b/conftest.py @@ -2,10 +2,11 @@ from __future__ import annotations +import io import uuid +from collections.abc import Generator from doctest import ELLIPSIS from pathlib import Path -from typing import TYPE_CHECKING import pytest from mock_vws import MockVWS @@ -17,10 +18,6 @@ PythonCodeBlockParser, ) -if TYPE_CHECKING: - import io - from collections.abc import Generator - @pytest.fixture(name="make_image_file") def fixture_make_image_file( diff --git a/pyproject.toml b/pyproject.toml index bb284945c..627d29142 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -165,7 +165,12 @@ lint.flake8-tidy-imports.banned-api."operator.attrgetter".msg = "operator.attrge lint.flake8-tidy-imports.banned-api."typing.cast".msg = """\ typing.cast is banned: use explicit type narrowing or a typed variable instead.\ """ -lint.flake8-type-checking.runtime-evaluated-decorators = [ "beartype.beartype" ] +# pytest-beartype evaluates fixture annotations at runtime. +lint.flake8-type-checking.runtime-evaluated-decorators = [ + "beartype.beartype", + "pytest.fixture", + "pytest_asyncio.fixture", +] lint.pydocstyle.convention = "google" lint.preview = true diff --git a/tests/async_vws/test_configuration.py b/tests/async_vws/test_configuration.py index d873f563a..591586a79 100644 --- a/tests/async_vws/test_configuration.py +++ b/tests/async_vws/test_configuration.py @@ -2,12 +2,13 @@ from __future__ import annotations -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import AsyncVWS if TYPE_CHECKING: @@ -15,12 +16,20 @@ @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_custom_base_url( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to add a target to a database under a custom VWS URL. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) base_vws_url = "http://example.com" with MockVWS(base_vws_url=base_vws_url) as mock: database = CloudDatabase() diff --git a/tests/async_vws/test_processing.py b/tests/async_vws/test_processing.py index c4014d8e5..c42d1b39c 100644 --- a/tests/async_vws/test_processing.py +++ b/tests/async_vws/test_processing.py @@ -2,12 +2,13 @@ from __future__ import annotations -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import AsyncVWS from vws.exceptions.custom_exceptions import ( TargetProcessingTimeoutError, @@ -21,12 +22,20 @@ @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_wait_for_target_processed( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to wait until a target is processed.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -48,10 +57,18 @@ async def test_wait_for_target_processed( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_custom_timeout( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to set a maximum timeout.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with MockVWS(processing_time_seconds=0.5) as mock: database = CloudDatabase() mock.add_cloud_database(cloud_database=database) diff --git a/tests/async_vws/test_reports.py b/tests/async_vws/test_reports.py index eac3ceb13..ca3e2edd2 100644 --- a/tests/async_vws/test_reports.py +++ b/tests/async_vws/test_reports.py @@ -3,15 +3,17 @@ from __future__ import annotations import calendar +import datetime import time import uuid from http import HTTPStatus -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import AsyncVWS from vws.exceptions.custom_exceptions import ( DatabaseIdNotSetError, @@ -29,7 +31,6 @@ from vws.response import Response if TYPE_CHECKING: - import datetime import io @@ -65,14 +66,22 @@ async def __call__( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_get_target_summary_report( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of a target are returned by ``get_target_summary_report``. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_name = uuid.uuid4().hex target_id = await async_vws_client.add_target( name=target_name, @@ -105,12 +114,21 @@ async def test_get_target( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames="report_period", argvalues=["current", "previous"] +) async def test_reco_counts_report( *, async_vws_client: AsyncVWS, - report_month: datetime.date, + current_month: datetime.date, + report_period: Literal["current", "previous"], ) -> None: """A report can be requested, waited for and downloaded.""" + report_month = ( + current_month + if report_period == "current" + else current_month - datetime.timedelta(days=1) + ) client = async_vws_client report_request = await client.request_database_reco_counts_report( year=report_month.year, diff --git a/tests/async_vws/test_targets.py b/tests/async_vws/test_targets.py index ff16890a0..53308cc35 100644 --- a/tests/async_vws/test_targets.py +++ b/tests/async_vws/test_targets.py @@ -7,11 +7,12 @@ import base64 import uuid -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest # pytest-beartype resolves fixture annotations at runtime. +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import AsyncCloudRecoService, AsyncVWS # noqa: TC001 from vws.reports import ( TargetRecord, @@ -31,15 +32,23 @@ argnames="active_flag", argvalues=[True, False], ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_add_target( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], application_metadata: bytes | None, async_cloud_reco_client: AsyncCloudRecoService, active_flag: bool, ) -> None: """No exception is raised when adding one target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) name = "x" width = 1 if application_metadata is None: @@ -84,10 +93,15 @@ async def test_add_target( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_add_two_targets( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """No exception is raised when adding two targets with different names. @@ -95,6 +109,9 @@ async def test_add_two_targets( This demonstrates that the image seek position is not changed. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) for name in ("a", "b"): await async_vws_client.add_target( name=name, @@ -106,12 +123,20 @@ async def test_add_two_targets( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_list_targets( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to get a list of target IDs.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) id_1 = await async_vws_client.add_target( name="x", width=1, @@ -131,12 +156,20 @@ async def test_list_targets( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_delete_target( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to delete a target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -158,14 +191,22 @@ async def test_delete_target( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_get_target_record( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of a target are returned by ``get_target_record``. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -191,12 +232,20 @@ async def test_get_target_record( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_get_duplicate_targets( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to get the IDs of similar targets.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -225,14 +274,22 @@ async def test_get_duplicate_targets( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_update_target( *, async_vws_client: AsyncVWS, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], different_high_quality_image: io.BytesIO, ) -> None: """It is possible to update a target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -276,12 +333,20 @@ async def test_update_target( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_no_fields_given( *, async_vws_client: AsyncVWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to give no update fields.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, diff --git a/tests/conftest.py b/tests/conftest.py index f6424bac5..119dbf98c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,7 +3,10 @@ from __future__ import annotations import datetime -from typing import TYPE_CHECKING, BinaryIO, Literal +import io +from collections.abc import AsyncGenerator, Generator +from contextlib import ExitStack +from pathlib import Path import pytest import pytest_asyncio @@ -11,6 +14,7 @@ from mock_vws.database import CloudDatabase, VuMarkDatabase from mock_vws.target import VuMarkTarget +from tests.image_factory import ImageFactory from vws import ( VWS, AsyncCloudRecoService, @@ -28,22 +32,12 @@ ModelTargetView, ) -if TYPE_CHECKING: - import io - from collections.abc import AsyncGenerator, Generator - from pathlib import Path - # The mock accepts one hard-coded pair of Model Target Web API OAuth2 # credentials, which it does not expose. _MODEL_TARGET_CLIENT_ID = "client-id" _MODEL_TARGET_CLIENT_CREDENTIALS = ("client-id", "client-secret") -def _image_file_mode(*, value: Literal["r+b", "rb"]) -> Literal["r+b", "rb"]: - """Type a file mode supplied by the pytest parameter API.""" - return value - - @pytest.fixture(name="_mock_database") def fixture_mock_database() -> Generator[CloudDatabase]: """Yield a mock ``CloudDatabase``.""" @@ -209,44 +203,16 @@ def fixture_current_month() -> datetime.date: return now.date().replace(day=1) -@pytest.fixture(name="report_month", params=["current", "previous"]) -def fixture_report_month(*, request: pytest.FixtureRequest) -> datetime.date: - """A month which a reco counts report can be requested for. - - Vuforia accepts only the current month and the previous month. - """ - now = datetime.datetime.now(tz=datetime.UTC) - first_of_month = now.date().replace(day=1) - if request.param == "current": - return first_of_month - - return first_of_month - datetime.timedelta(days=1) - - -@pytest.fixture(name="image_file", params=["r+b", "rb"]) -def fixture_image_file( +@pytest.fixture(name="image_factory") +def fixture_image_factory( *, high_quality_image: io.BytesIO, tmp_path: Path, - request: pytest.FixtureRequest, -) -> Generator[BinaryIO]: - """An image file object.""" - file = tmp_path / "image.jpg" - buffer = high_quality_image.getvalue() - _ = file.write_bytes(data=buffer) - mode = _image_file_mode(value=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, - high_quality_image: io.BytesIO, - image_file: BinaryIO, -) -> io.BytesIO | BinaryIO: - """An image in any of the types that the API accepts.""" - if request.param == "high_quality_image": - return high_quality_image - return image_file +) -> Generator[ImageFactory]: + """Create selected image inputs with scoped file-handle cleanup.""" + with ExitStack() as stack: + yield ImageFactory( + high_quality_image=high_quality_image, + directory=tmp_path, + stack=stack, + ) diff --git a/tests/image_factory.py b/tests/image_factory.py new file mode 100644 index 000000000..32a391af2 --- /dev/null +++ b/tests/image_factory.py @@ -0,0 +1,55 @@ +"""Typed image cases and resources for API tests.""" + +from __future__ import annotations + +import io +from typing import TYPE_CHECKING, BinaryIO, Literal + +from beartype import beartype + +if TYPE_CHECKING: + from contextlib import ExitStack + from pathlib import Path + +# Keep the buffer case for each file mode, matching the original nested +# image and image_file fixture cases. +IMAGE_CASES: tuple[ + tuple[Literal["high_quality_image", "image_file"], Literal["r+b", "rb"]], + ..., +] = ( + ("high_quality_image", "r+b"), + ("high_quality_image", "rb"), + ("image_file", "r+b"), + ("image_file", "rb"), +) + + +class ImageFactory: + """Create image inputs whose file handles close with the fixture.""" + + def __init__( + self, + *, + high_quality_image: io.BytesIO, + directory: Path, + stack: ExitStack, + ) -> None: + """Keep the image bytes, temporary directory and cleanup stack.""" + self._high_quality_image = high_quality_image + self._directory = directory + self._stack = stack + + @beartype + def create( + self, + *, + source: Literal["high_quality_image", "image_file"], + mode: Literal["r+b", "rb"], + ) -> io.BytesIO | BinaryIO: + """Open the file and return the selected image input.""" + file = self._directory / "image.jpg" + _ = file.write_bytes(data=self._high_quality_image.getvalue()) + image_file = self._stack.enter_context(cm=file.open(mode=mode)) + if source == "high_quality_image": + return self._high_quality_image + return image_file diff --git a/tests/test_async_query.py b/tests/test_async_query.py index 200a4d4fa..0c067faa3 100644 --- a/tests/test_async_query.py +++ b/tests/test_async_query.py @@ -6,12 +6,13 @@ import secrets import uuid from http import HTTPStatus -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import AsyncCloudRecoService, AsyncVWS from vws.exceptions.custom_exceptions import ServerError from vws.include_target_data import CloudRecoIncludeTargetData @@ -52,12 +53,20 @@ async def __call__( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_no_matches( *, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """An empty list is returned if there are no matches.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) result = await async_cloud_reco_client.query( image=image, ) @@ -65,13 +74,21 @@ async def test_no_matches( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_match( *, async_vws_client: AsyncVWS, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of matching targets are returned.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name="x", width=1, @@ -89,8 +106,19 @@ async def test_match( @pytest.mark.asyncio -async def test_server_error(*, image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +async def test_server_error( + *, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """Server errors are exposed through the public async client.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) client = AsyncCloudRecoService( client_access_key="access-key", client_secret_key=secrets.token_hex(), @@ -102,12 +130,20 @@ async def test_server_error(*, image: io.BytesIO | BinaryIO) -> None: @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_custom_base_url( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to query a target in a database under a custom VWQ URL. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) base_vwq_url = "http://example.com" with MockVWS(base_vwq_url=base_vwq_url) as mock: database = CloudDatabase() @@ -144,15 +180,23 @@ async def test_custom_base_url( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_custom( *, async_vws_client: AsyncVWS, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to set a custom ``max_num_results``. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -192,15 +236,23 @@ async def test_custom( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_none( *, async_vws_client: AsyncVWS, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """When ``CloudRecoIncludeTargetData.NONE`` is given, target data is not returned in any match. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -219,15 +271,23 @@ async def test_none( @pytest.mark.asyncio +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) async def test_all( *, async_vws_client: AsyncVWS, async_cloud_reco_client: AsyncCloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """When ``CloudRecoIncludeTargetData.ALL`` is given, target data is returned in all matches. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = await async_vws_client.add_target( name=uuid.uuid4().hex, width=1, diff --git a/tests/test_query.py b/tests/test_query.py index e0a4309bd..a154cc425 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -7,7 +7,7 @@ import secrets import uuid from http import HTTPStatus -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest import requests @@ -15,6 +15,7 @@ from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import VWS, CloudRecoService from vws.exceptions.custom_exceptions import ServerError from vws.include_target_data import CloudRecoIncludeTargetData @@ -63,23 +64,39 @@ def __call__( ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_no_matches( *, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """An empty list is returned if there are no matches.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) result = cloud_reco_client.query(image=image) assert result == [] +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_match( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of matching targets are returned.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1, @@ -92,8 +109,19 @@ def test_match( assert matching_target.target_id == target_id -def test_invalid_results(*, image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +def test_invalid_results( + *, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """Query results in responses must be a list of objects.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) transport = _JSONResponseTransport( body={"result_code": "Success", "results": 1}, status_code=HTTPStatus.OK, @@ -108,8 +136,19 @@ def test_invalid_results(*, image: io.BytesIO | BinaryIO) -> None: _ = client.query(image=image) -def test_server_error(*, image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +def test_server_error( + *, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """Server errors are exposed through the public query client.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) transport = _JSONResponseTransport( body={}, status_code=HTTPStatus.INTERNAL_SERVER_ERROR, @@ -128,15 +167,23 @@ def test_server_error(*, image: io.BytesIO | BinaryIO) -> None: argnames=("response_delay_seconds", "expect_timeout"), argvalues=[(29, False), (31, True)], ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_default_timeout( *, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], response_delay_seconds: int, expect_timeout: bool, ) -> None: """At 29 seconds there is no error; at 31 seconds there is a timeout. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with ( freeze_time() as frozen_datetime, MockVWS( @@ -179,14 +226,22 @@ def test_default_timeout( ((5.0, 0.1), 0.11, True), ], ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_custom_timeout( *, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], custom_timeout: float | tuple[float, float], response_delay_seconds: float, expect_timeout: bool, ) -> None: """Custom timeouts are honored for both float and tuple forms.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with ( freeze_time() as frozen_datetime, MockVWS( @@ -217,13 +272,23 @@ def test_custom_timeout( assert not bool(matches) -def test_custom_base_url(image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +def test_custom_base_url( + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """ It is possible to use query a target to a database under a custom VWQ URL. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) base_vwq_url = "http://example.com" with MockVWS(base_vwq_url=base_vwq_url) as mock: database = CloudDatabase() @@ -255,13 +320,21 @@ def test_custom_base_url(image: io.BytesIO | BinaryIO) -> None: assert match.target_id == target_id +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_custom_base_url_with_path_prefix( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """ A base VWQ URL with a path prefix is used as-is, without the prefix being dropped. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) base_vwq_url = "http://example.com/prefix" with MockVWS(base_vwq_url=base_vwq_url) as mock: database = CloudDatabase() @@ -288,13 +361,21 @@ def test_custom_base_url_with_path_prefix( assert len(matches) == 1 +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_max_num_results_default( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """By default the maximum number of results is 1.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -315,13 +396,21 @@ def test_max_num_results_default( assert len(matches) == 1 +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_custom( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to set a custom ``max_num_results``.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -354,13 +443,21 @@ def test_custom( assert len(matches) == max_num_results +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_include_target_data_default( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """By default, target data is only returned in the top match.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -385,17 +482,25 @@ def test_include_target_data_default( assert second_match.target_data is None +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_top( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """ When ``CloudRecoIncludeTargetData.TOP`` is given, target data is only returned in the top match. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -421,11 +526,16 @@ def test_top( assert second_match.target_data is None +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_none( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """ When ``CloudRecoIncludeTargetData.NONE`` is given, target data @@ -433,6 +543,9 @@ def test_none( not returned in any match. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, @@ -458,16 +571,24 @@ def test_none( assert second_match.target_data is None +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_all( *, vws_client: VWS, cloud_reco_client: CloudRecoService, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """ When ``CloudRecoIncludeTargetData.ALL`` is given, target data is returned in all matches. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name=uuid.uuid4().hex, width=1, diff --git a/tests/vws/test_configuration.py b/tests/vws/test_configuration.py index b510e61da..d9e4094fe 100644 --- a/tests/vws/test_configuration.py +++ b/tests/vws/test_configuration.py @@ -3,7 +3,7 @@ from __future__ import annotations import datetime -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest import requests @@ -11,6 +11,7 @@ from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import VWS if TYPE_CHECKING: @@ -21,15 +22,23 @@ argnames=("response_delay_seconds", "expect_timeout"), argvalues=[(29, False), (31, True)], ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_default_timeout( *, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], response_delay_seconds: int, expect_timeout: bool, ) -> None: """At 29 seconds there is no error; at 31 seconds there is a timeout. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with ( freeze_time() as frozen_datetime, MockVWS( @@ -83,14 +92,22 @@ def test_default_timeout( ((5.0, 0.1), 0.11, True), ], ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_custom_timeout( *, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], custom_timeout: float | tuple[float, float], response_delay_seconds: float, expect_timeout: bool, ) -> None: """Custom timeouts are honored for both float and tuple forms.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with ( freeze_time() as frozen_datetime, MockVWS( @@ -132,12 +149,22 @@ def test_custom_timeout( ) -def test_custom_base_url(image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +def test_custom_base_url( + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """ It is possible to use add a target to a database under a custom VWS URL. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) base_vws_url = "http://example.com" with MockVWS(base_vws_url=base_vws_url) as mock: database = CloudDatabase() diff --git a/tests/vws/test_processing.py b/tests/vws/test_processing.py index 0c30fdc1d..69af3cc58 100644 --- a/tests/vws/test_processing.py +++ b/tests/vws/test_processing.py @@ -2,12 +2,13 @@ from __future__ import annotations -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import VWS from vws.exceptions.custom_exceptions import ( TargetProcessingTimeoutError, @@ -20,12 +21,20 @@ import io +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_wait_for_target_processed( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to wait until a target is processed.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1, @@ -40,10 +49,18 @@ def test_wait_for_target_processed( assert report.status != TargetStatuses.PROCESSING +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_default_seconds_between_requests( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """By default, 0.2 seconds are waited between polling requests.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with MockVWS(processing_time_seconds=0.5) as mock: database = CloudDatabase() mock.add_cloud_database(cloud_database=database) @@ -90,12 +107,20 @@ def test_default_seconds_between_requests( assert report.request_usage == expected_requests +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_custom_seconds_between_requests( - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to customize the time waited between polling requests. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with MockVWS(processing_time_seconds=0.5) as mock: database = CloudDatabase() mock.add_cloud_database(cloud_database=database) @@ -142,8 +167,18 @@ def test_custom_seconds_between_requests( assert report.request_usage == expected_requests -def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) +def test_custom_timeout( + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], +) -> None: """It is possible to set a maximum timeout.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) with MockVWS(processing_time_seconds=0.5) as mock: database = CloudDatabase() mock.add_cloud_database(cloud_database=database) diff --git a/tests/vws/test_reports.py b/tests/vws/test_reports.py index 87885efa5..8f2ad0368 100644 --- a/tests/vws/test_reports.py +++ b/tests/vws/test_reports.py @@ -7,13 +7,14 @@ import time import uuid from http import HTTPStatus -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest from freezegun import freeze_time from mock_vws import MockVWS from mock_vws.database import CloudDatabase +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import VWS from vws.exceptions.custom_exceptions import ( DatabaseIdNotSetError, @@ -69,14 +70,22 @@ def __call__( ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_get_target_summary_report( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of a target are returned by ``get_target_summary_report``. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) date = "2018-04-25" target_name = uuid.uuid4().hex with freeze_time(time_to_freeze=date): @@ -152,12 +161,21 @@ def test_get_target(vws_client: VWS) -> None: assert report == expected_report +@pytest.mark.parametrize( + argnames="report_period", argvalues=["current", "previous"] +) def test_reco_counts_report( *, vws_client: VWS, - report_month: datetime.date, + current_month: datetime.date, + report_period: Literal["current", "previous"], ) -> None: """A report can be requested, waited for and downloaded.""" + report_month = ( + current_month + if report_period == "current" + else current_month - datetime.timedelta(days=1) + ) report_request = vws_client.request_database_reco_counts_report( year=report_month.year, month=calendar.Month(value=report_month.month), diff --git a/tests/vws/test_targets.py b/tests/vws/test_targets.py index 58190c65d..2ee5427ed 100644 --- a/tests/vws/test_targets.py +++ b/tests/vws/test_targets.py @@ -10,10 +10,11 @@ import secrets import uuid from http import HTTPStatus -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Literal import pytest +from tests.image_factory import IMAGE_CASES, ImageFactory from vws import VWS, CloudRecoService from vws.reports import ( TargetRecord, @@ -63,15 +64,23 @@ def __call__( argvalues=[None, b"a"], ) @pytest.mark.parametrize(argnames="active_flag", argvalues=[True, False]) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_add_target( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], application_metadata: bytes | None, cloud_reco_client: CloudRecoService, active_flag: bool, ) -> None: """No exception is raised when adding one target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) name = "x" width = 1 if application_metadata is None: @@ -105,16 +114,24 @@ def test_add_target( assert matching_targets == [] +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_add_two_targets( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """No exception is raised when adding two targets with different names. This demonstrates that the image seek position is not changed. """ + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) for name in ("a", "b"): _ = vws_client.add_target( name=name, @@ -125,12 +142,20 @@ def test_add_two_targets( ) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_list_targets( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to get a list of target IDs.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) id_1 = vws_client.add_target( name="x", width=1, @@ -148,12 +173,20 @@ def test_list_targets( assert sorted(vws_client.list_targets()) == sorted([id_1, id_2]) +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_delete_target( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to delete a target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1, @@ -168,12 +201,20 @@ def test_delete_target( assert vws_client.list_targets() == [] +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_get_target_record( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """Details of a target are returned by ``get_target_record``.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1, @@ -231,12 +272,20 @@ def test_get_failed( assert result.status == TargetStatuses.FAILED +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_get_duplicate_targets( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to get the IDs of similar targets.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1, @@ -297,14 +346,22 @@ def test_invalid_vws_response_envelope(*, body: object) -> None: client.delete_target(target_id="target-id") +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_update_target( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], different_high_quality_image: io.BytesIO, cloud_reco_client: CloudRecoService, ) -> None: """It is possible to update a target.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) old_name = uuid.uuid4().hex old_width = secrets.choice(seq=range(1, 5000)) / 100 target_id = vws_client.add_target( @@ -357,12 +414,20 @@ def test_update_target( assert not target_details.target_record.active_flag +@pytest.mark.parametrize( + argnames=("image_source", "image_mode"), argvalues=IMAGE_CASES +) def test_no_fields_given( *, vws_client: VWS, - image: io.BytesIO | BinaryIO, + image_factory: ImageFactory, + image_source: Literal["high_quality_image", "image_file"], + image_mode: Literal["r+b", "rb"], ) -> None: """It is possible to give no update fields.""" + image: io.BytesIO | BinaryIO = image_factory.create( + source=image_source, mode=image_mode + ) target_id = vws_client.add_target( name="x", width=1,