From b440a11e96a589cfca5f0dace79830f41544ca60 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 16 Aug 2026 12:41:51 +0100 Subject: [PATCH] Isolate each call of a function decorated with a MockVWS instance `MockVWS` inherited from `ContextDecorator`, so every use of one instance shared a single set of databases and targets. Decorating two test functions with one instance therefore made them affect each other, which is the opposite of what the decorator idiom implies and of what `@responses.activate` and `@mock.patch` do. Each call of a decorated function now gets its own target manager and its own API fakes. The databases added to the instance are available inside the call, and, because targets are stored on the caller's own database object, their targets are snapshotted and restored around the call so that a database can still be inspected while the call runs. Using an instance as a context manager is unchanged: a `with` block still shares its state with every other use of the same instance. Along the way: - Replace the constructor arguments held for this purpose with a frozen `_MockVWSOptions` dataclass, so the options are read as typed attributes rather than unpacked from a `dict[str, Any]`. - Keep the started mocks as a stack, so that a decorated function calling another decorated function no longer leaves the outer call's mocking stopped. Co-Authored-By: Claude Opus 5 (1M context) --- docs/source/basic-example.rst | 26 +++ newsfragments/mock-vws-decorator.change | 4 + src/mock_vws/decorators.py | 238 ++++++++++++++++++--- tests/mock_vws/test_requests_mock_usage.py | 195 +++++++++++++++-- 4 files changed, 414 insertions(+), 49 deletions(-) create mode 100644 newsfragments/mock-vws-decorator.change diff --git a/docs/source/basic-example.rst b/docs/source/basic-example.rst index c6829a3b4..281cd20af 100644 --- a/docs/source/basic-example.rst +++ b/docs/source/basic-example.rst @@ -17,6 +17,32 @@ By default, an exception will be raised if any requests to unmocked addresses are made. +A ``MockVWS`` instance can also decorate a function: + +.. code-block:: python + + """Make a request to the Vuforia mock from a decorated function.""" + + import requests + + from mock_vws import MockVWS + from mock_vws.database import CloudDatabase + + mock = MockVWS() + mock.add_cloud_database(cloud_database=CloudDatabase()) + + + @mock + def get_summary() -> None: + """Make a request which uses the Vuforia mock.""" + requests.get(url="https://vws.vuforia.com/summary", timeout=30) + + + get_summary() + +Each call of a decorated function gets its own databases and targets, so decorated functions do not affect each other. +A ``with`` block, by contrast, shares one set of databases and targets with every other use of the same instance. + See :ref:`mock-api-reference` for details of what can be changed and how. .. _requests: https://pypi.org/project/requests/ diff --git a/newsfragments/mock-vws-decorator.change b/newsfragments/mock-vws-decorator.change new file mode 100644 index 000000000..15c66b889 --- /dev/null +++ b/newsfragments/mock-vws-decorator.change @@ -0,0 +1,4 @@ +Give each call of a function decorated with a ``MockVWS`` instance its own databases and targets. +Previously, every use of an instance shared one set of databases and targets, so decorating two test functions with one instance made them affect each other. +A database can still be inspected during a call, and its targets are what they were before the call again once it returns. +Using an instance as a context manager is unchanged: a ``with`` block still shares its state with every other use of the same instance. diff --git a/src/mock_vws/decorators.py b/src/mock_vws/decorators.py index e69972d1d..1eee47d9a 100644 --- a/src/mock_vws/decorators.py +++ b/src/mock_vws/decorators.py @@ -1,9 +1,11 @@ """Decorators for using the mock.""" +import functools import re import time -from collections.abc import Callable, Mapping -from contextlib import ContextDecorator +from collections.abc import Callable, Generator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal, Self from urllib.parse import urlparse @@ -49,10 +51,45 @@ @beartype(conf=BeartypeConf(is_pep484_tower=True)) -class MockVWS(ContextDecorator): +@dataclass(eq=True, frozen=True, kw_only=True) +class _MockVWSOptions: + """The options which configure a mock. + + These are everything a mock is given when it is created, as opposed to + the databases and targets which it accumulates as it is used. + """ + + base_vws_url: str + base_vwq_url: str + cloud_query_failure_response: CloudQueryFailureResponse | None + duplicate_match_checker: ImageMatcher + query_match_checker: ImageMatcher + processing_time_seconds: float + model_target_generation_failure: ModelTargetGenerationFailure | None + model_target_generation_warning: ModelTargetGenerationWarning | None + target_tracking_rater: TargetTrackingRater + real_http: bool + response_delay_seconds: float + sleep_fn: Callable[[float], None] + vumark_generation_failure: VuMarkGenerationFailure | None + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class MockVWS: """Route requests to Vuforia's Web Service APIs to fakes of those APIs. Works with both ``requests`` and ``httpx``. + + An instance is usable as a context manager and as a decorator. + + A context manager block shares one set of databases and targets with + every other use of the same instance, so state created in one ``with`` + block is still there in the next one. + + A decorated function instead gets its own databases and targets for the + duration of each call. The databases added to the instance are available + inside the call, and the targets created during the call are discarded + when it returns, so decorated functions do not affect each other. """ def __init__( @@ -123,7 +160,6 @@ def __init__( ValueError: Both a Model Target generation failure and warning are configured. """ - super().__init__() if ( model_target_generation_failure is not None and model_target_generation_warning is not None @@ -133,36 +169,109 @@ def __init__( "are mutually exclusive" ) raise ValueError(msg) - self._real_http = real_http - self._response_delay_seconds = response_delay_seconds - self._sleep_fn = sleep_fn - self._mock: RequestsMock - self._router: respx.MockRouter - self._target_manager = TargetManager() - self._base_vws_url = base_vws_url - self._base_vwq_url = base_vwq_url for url in (base_vwq_url, base_vws_url): parse_result = urlparse(url=url) if not parse_result.scheme: raise MissingSchemeError(url=url) - self._mock_vws_api = MockVuforiaWebServicesAPI( - target_manager=self._target_manager, + # The options are kept so that decorating a function can build an + # equivalently configured set of fakes, with their own databases and + # targets, for each call of that function. + self._options = _MockVWSOptions( base_vws_url=base_vws_url, + base_vwq_url=base_vwq_url, + cloud_query_failure_response=cloud_query_failure_response, + duplicate_match_checker=duplicate_match_checker, + query_match_checker=query_match_checker, processing_time_seconds=float(processing_time_seconds), model_target_generation_failure=model_target_generation_failure, model_target_generation_warning=model_target_generation_warning, - duplicate_match_checker=duplicate_match_checker, target_tracking_rater=target_tracking_rater, + real_http=real_http, + response_delay_seconds=response_delay_seconds, + sleep_fn=sleep_fn, vumark_generation_failure=vumark_generation_failure, ) + # A mock can be started while it is already started, for example + # when a decorated function calls another decorated function, so the + # started mocks are kept as a stack. + self._started: list[tuple[RequestsMock, respx.MockRouter]] = [] + self._added_cloud_databases: list[CloudDatabase] = [] + self._added_vumark_databases: list[VuMarkDatabase] = [] + self._target_manager = TargetManager() + self._mock_vws_api, self._mock_vwq_api = self._build_apis( + target_manager=self._target_manager, + ) - self._mock_vwq_api = MockVuforiaWebQueryAPI( + def _build_apis( + self, + *, + target_manager: TargetManager, + ) -> tuple[MockVuforiaWebServicesAPI, MockVuforiaWebQueryAPI]: + """Build fakes of the Vuforia APIs, backed by a target manager. + + Args: + target_manager: The target manager which the fakes use. + + Returns: + A fake of the VWS API and a fake of the VWQ API. + """ + options = self._options + mock_vws_api = MockVuforiaWebServicesAPI( + target_manager=target_manager, + base_vws_url=options.base_vws_url, + processing_time_seconds=options.processing_time_seconds, + model_target_generation_failure=( + options.model_target_generation_failure + ), + model_target_generation_warning=( + options.model_target_generation_warning + ), + duplicate_match_checker=options.duplicate_match_checker, + target_tracking_rater=options.target_tracking_rater, + vumark_generation_failure=options.vumark_generation_failure, + ) + mock_vwq_api = MockVuforiaWebQueryAPI( + target_manager=target_manager, + query_match_checker=options.query_match_checker, + failure_response=options.cloud_query_failure_response, + ) + return mock_vws_api, mock_vwq_api + + @contextmanager + def _fresh_state(self) -> Generator[None]: + """Swap in databases and targets which are used only in this block. + + The databases added to this instance are added to the new state, and + the state which was there before the block is back once it ends. + + Yields: + ``None``. + """ + original_target_manager = self._target_manager + original_mock_vws_api = self._mock_vws_api + original_mock_vwq_api = self._mock_vwq_api + + self._target_manager = TargetManager() + self._mock_vws_api, self._mock_vwq_api = self._build_apis( target_manager=self._target_manager, - query_match_checker=query_match_checker, - failure_response=cloud_query_failure_response, ) + for cloud_database in self._added_cloud_databases: + self._target_manager.add_cloud_database( + cloud_database=cloud_database, + ) + for vumark_database in self._added_vumark_databases: + self._target_manager.add_vumark_database( + vumark_database=vumark_database, + ) + + try: + yield + finally: + self._target_manager = original_target_manager + self._mock_vws_api = original_mock_vws_api + self._mock_vwq_api = original_mock_vwq_api def add_cloud_database(self, cloud_database: CloudDatabase) -> None: """Add a cloud database. @@ -177,6 +286,7 @@ def add_cloud_database(self, cloud_database: CloudDatabase) -> None: self._target_manager.add_cloud_database( cloud_database=cloud_database, ) + self._added_cloud_databases.append(cloud_database) def add_vumark_database(self, vumark_database: VuMarkDatabase) -> None: """Add a VuMark database. @@ -191,6 +301,67 @@ def add_vumark_database(self, vumark_database: VuMarkDatabase) -> None: self._target_manager.add_vumark_database( vumark_database=vumark_database, ) + self._added_vumark_databases.append(vumark_database) + + def __call__[**P, T]( + self, + function: Callable[P, T], + ) -> Callable[P, T]: + """Wrap a function so that each call of it runs against the mock. + + Each call gets its own databases and targets, so that decorated + functions do not affect each other. The databases added to this + instance are available inside the call, and their targets are what + they were before the call again once it returns. + + Args: + function: The function to wrap. + + Returns: + The wrapped function. + """ + + @functools.wraps(wrapped=function) + def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: + """Run the given function against a mock of its own. + + Returns: + The return value of the given function. + """ + # The targets of a database are stored on the database object + # itself, and that object belongs to the caller, so a new target + # manager is not enough to isolate one call from the next. We + # therefore put the targets back as they were afterwards. + # + # ``CloudDatabase`` equality includes the targets, which change + # during the call, so the snapshots are held in a list rather + # than in a dictionary keyed by database. + # + # Reading and writing the targets in a database is guarded by + # the target manager's lock, as documented on that lock. + with self._target_manager.lock: + cloud_snapshots = [ + (database, set(database.targets)) + for database in self._added_cloud_databases + ] + vumark_snapshots = [ + (database, set(database.vumark_targets)) + for database in self._added_vumark_databases + ] + + try: + with self._fresh_state(), self: + return function(*args, **kwargs) + finally: + with self._target_manager.lock: + for cloud_database, cloud_targets in cloud_snapshots: + cloud_database.targets.clear() + cloud_database.targets.update(cloud_targets) + for vumark_database, vumark_targets in vumark_snapshots: + vumark_database.vumark_targets.clear() + vumark_database.vumark_targets.update(vumark_targets) + + return wrapper @staticmethod def _wrap_callback( @@ -264,8 +435,8 @@ def __enter__(self) -> Self: mock = RequestsMock(assert_all_requests_are_fired=False) for api, base_url in ( - (self._mock_vws_api, self._base_vws_url), - (self._mock_vwq_api, self._base_vwq_url), + (self._mock_vws_api, self._options.base_vws_url), + (self._mock_vwq_api, self._options.base_vwq_url), ): base_path = urlparse(url=base_url).path.rstrip("/") for route in api.routes: @@ -282,30 +453,30 @@ def __enter__(self) -> Self: url=compiled_url_pattern, callback=self._wrap_callback( callback=original_callback, - delay_seconds=self._response_delay_seconds, - sleep_fn=self._sleep_fn, + delay_seconds=self._options.response_delay_seconds, + sleep_fn=self._options.sleep_fn, base_path=base_path, ), content_type=None, ) - if self._real_http: + if self._options.real_http: all_requests_pattern = re.compile(pattern=".*") mock.add_passthru(prefix=all_requests_pattern) - self._mock = mock - self._mock.start() + mock.start() - self._router = start_respx_router( + router = start_respx_router( mock_vws_api=self._mock_vws_api, mock_vwq_api=self._mock_vwq_api, - base_vws_url=self._base_vws_url, - base_vwq_url=self._base_vwq_url, - response_delay_seconds=self._response_delay_seconds, - sleep_fn=self._sleep_fn, - real_http=self._real_http, + base_vws_url=self._options.base_vws_url, + base_vwq_url=self._options.base_vwq_url, + response_delay_seconds=self._options.response_delay_seconds, + sleep_fn=self._options.sleep_fn, + real_http=self._options.real_http, ) + self._started.append((mock, router)) return self def __exit__(self, *exc: object) -> Literal[False]: @@ -318,6 +489,7 @@ def __exit__(self, *exc: object) -> Literal[False]: # unused, so we "use" it here. del exc - self._mock.stop() - self._router.stop() + mock, router = self._started.pop() + mock.stop() + router.stop() return False diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index 18e77d960..98d80685a 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -1393,6 +1393,38 @@ def test_duplicate_vumark_keys() -> None: mock.add_vumark_database(vumark_database=bad_database) +class TestContextManagerReuse: + """Tests for reusing a ``MockVWS`` instance as a context manager.""" + + @staticmethod + def test_state_is_kept_between_uses( + high_quality_image: io.BytesIO, + ) -> None: + """A ``MockVWS`` instance keeps its databases, and the targets in + them, between ``with`` blocks. + """ + database = CloudDatabase() + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + mock = MockVWS() + mock.add_cloud_database(cloud_database=database) + + with mock: + vws_client.add_target( + name="my-target", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + assert len(vws_client.list_targets()) == 1 + + with mock: + assert len(vws_client.list_targets()) == 1 + + class TestQueryImageMatchers: """Tests for query image matchers.""" @@ -1984,38 +2016,169 @@ def add_target() -> TargetStatuses: assert add_target() == TargetStatuses.FAILED @staticmethod - def test_targets_persist_between_calls( + def test_each_call_is_isolated(high_quality_image: io.BytesIO) -> None: + """Each call of a decorated function has its own targets. + + Targets created by one call are not there in the next call, or in a + call of another function decorated with the same instance. + """ + database = CloudDatabase() + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + mock = MockVWS(processing_time_seconds=0) + mock.add_cloud_database(cloud_database=database) + + @mock + def add_one_target() -> None: + """Add a target with a name used only once per call.""" + vws_client.add_target( + name="only-one", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + assert len(vws_client.list_targets()) == 1 + + @mock + def count_targets() -> int: + """Return the number of targets in the database.""" + return len(vws_client.list_targets()) + + add_one_target() + assert count_targets() == 0 + add_one_target() + assert count_targets() == 0 + + @staticmethod + def test_nested_calls(high_quality_image: io.BytesIO) -> None: + """A decorated function can call another decorated function. + + The inner call starts from the targets which are there when it is + called, and the targets it creates are gone once it has returned. The + outer call keeps its own targets and its mocking. + """ + database = CloudDatabase() + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + mock = MockVWS(processing_time_seconds=0) + mock.add_cloud_database(cloud_database=database) + + @mock + def add_inner_target() -> int: + """Add a target and return the number of targets. + + Returns: + The number of targets, including the one added here. + """ + vws_client.add_target( + name="inner", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + return len(vws_client.list_targets()) + + @mock + def add_outer_target() -> tuple[int, int]: + """Add a target and make an inner call. + + Returns: + The number of targets seen by the inner call, and the number + of targets seen here once the inner call has returned. + """ + vws_client.add_target( + name="outer", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + return add_inner_target(), len(vws_client.list_targets()) + + targets_seen_by_inner_call = 2 + inner_count, outer_count = add_outer_target() + assert inner_count == targets_seen_by_inner_call + assert outer_count == 1 + assert not database.targets + + @staticmethod + def test_database_targets_are_restored( high_quality_image: io.BytesIO, ) -> None: - """A mock instance keeps its targets between calls of a decorated - function. + """A database is restored to the targets it had before a call. + + The targets can be inspected during the call, and they are what they + were before the call again once it returns. """ database = CloudDatabase() + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) mock = MockVWS(processing_time_seconds=0) mock.add_cloud_database(cloud_database=database) @mock - def add_target_and_list_targets() -> list[str]: - """Add a target and return the identifiers of all targets.""" - vws_client = VWS( - server_access_key=database.server_access_key, - server_secret_key=database.server_secret_key, - ) + def add_one_target() -> None: + """Add a target and inspect it on the database object.""" vws_client.add_target( - name=f"example-{len(vws_client.list_targets())}", + name="only-one", width=1, image=high_quality_image, active_flag=True, application_metadata=None, ) - return vws_client.list_targets() + (target,) = database.targets + assert target.name == "only-one" - expected_targets_after_second_call = 2 - assert len(add_target_and_list_targets()) == 1 - assert ( - len(add_target_and_list_targets()) - == expected_targets_after_second_call + add_one_target() + assert not database.targets + + @staticmethod + def test_exception_restores_database_targets( + high_quality_image: io.BytesIO, + ) -> None: + """The targets of a database are restored even when the decorated + function raises. + """ + database = CloudDatabase() + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, ) + mock = MockVWS(processing_time_seconds=0) + mock.add_cloud_database(cloud_database=database) + + @mock + def add_one_target_then_raise() -> None: + """Add a target and then raise an exception. + + Raises: + ValueError: Always. + """ + vws_client.add_target( + name="only-one", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + message = "Something went wrong." + raise ValueError(message) + + with pytest.raises( + expected_exception=ValueError, + match=r"^Something went wrong\.$", + ): + add_one_target_then_raise() + + assert not database.targets @staticmethod def test_query(high_quality_image: io.BytesIO) -> None: