From 41b049cafcefaab668b759235f9e9ea97168c3aa Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 15 Aug 2026 23:11:53 +0100 Subject: [PATCH] Add tests for using the mock as a decorator Co-Authored-By: Claude Opus 5 (1M context) --- tests/mock_vws/test_requests_mock_usage.py | 236 +++++++++++++++++++++ 1 file changed, 236 insertions(+) diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index cc1586163..3bef9f2c7 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -24,6 +24,8 @@ TargetQuotaReachedError, TooManyRequestsError, ) +from vws.reports import TargetStatuses +from vws.transports import HTTPXTransport from vws_auth_tools import authorization_header, rfc_1123_date from mock_vws import MissingSchemeError, MockVWS @@ -83,6 +85,16 @@ def _not_exact_matcher( return first_image_content != second_image_content +@beartype +def _unused_local_url() -> str: + """Return a URL for a local address with nothing listening on it.""" + sock = socket.socket() + sock.bind(("", 0)) + port = sock.getsockname()[1] + sock.close() + return f"http://localhost:{port}" + + @beartype def request_unmocked_address() -> None: """Make a request, using `requests` to an unmocked, free local address. @@ -1831,3 +1843,227 @@ def test_bearer_token_required() -> None: "message": "no Bearer token", "target": "jwt", } + + +class TestDecorator: + """Tests for using the mock as a decorator.""" + + @staticmethod + def test_requests_are_mocked_only_within_the_function() -> None: + """Requests to Vuforia are mocked within the decorated function, + and + they are not mocked once the decorated function has returned. + """ + base_vws_url = _unused_local_url() + summary_url = base_vws_url + "/summary" + + @MockVWS(base_vws_url=base_vws_url) + def make_request() -> requests.Response: + """Make a request to the mocked VWS API.""" + return requests.get( + url=summary_url, + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + data=b"", + timeout=30, + ) + + response = make_request() + assert response.status_code == HTTPStatus.BAD_REQUEST + + # Nothing is listening on the given address, so this shows that the + # mocking stops when the decorated function returns. + with pytest.raises( + expected_exception=requests.exceptions.ConnectionError + ): + requests.get(url=summary_url, timeout=30) + + @staticmethod + def test_httpx_requests_are_mocked() -> None: + """Requests made with ``httpx`` are mocked within the decorated + function. + """ + base_vws_url = _unused_local_url() + summary_url = base_vws_url + "/summary" + + @MockVWS(base_vws_url=base_vws_url) + def make_request() -> httpx.Response: + """Make a request to the mocked VWS API.""" + return httpx.get( + url=summary_url, + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=30, + ) + + response = make_request() + assert response.status_code == HTTPStatus.BAD_REQUEST + + with pytest.raises(expected_exception=httpx.ConnectError): + httpx.get(url=summary_url, timeout=30) + + @staticmethod + def test_arguments_and_return_value() -> None: + """Arguments are passed to the decorated function, and its return + value is returned. + """ + + @MockVWS() + def join(*parts: str, separator: str) -> str: + """Join the given parts.""" + return separator.join(parts) + + assert join("a", "b", separator="-") == "a-b" + + @staticmethod + def test_function_metadata_is_preserved() -> None: + """The decorated function keeps its name and docstring.""" + + @MockVWS() + def my_function() -> None: + """My docstring.""" + + assert my_function.__name__ == "my_function" + assert my_function.__doc__ == "My docstring." + + @staticmethod + def test_databases_added_before_decorating() -> None: + """Databases added to the mock are available within the decorated + function. + """ + database = CloudDatabase() + mock = MockVWS() + mock.add_cloud_database(cloud_database=database) + + @mock + def get_database_name() -> str: + """Get the name of the database from the mock.""" + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + return vws_client.get_database_summary_report().name + + assert get_database_name() == database.database_name + + @staticmethod + def test_options_are_used(image_file_failed_state: io.BytesIO) -> None: + """Options given to the mock are used within the decorated + function. + """ + database = CloudDatabase() + mock = MockVWS(processing_time_seconds=0) + mock.add_cloud_database(cloud_database=database) + + @mock + def add_target() -> TargetStatuses: + """Add a target and return its status immediately.""" + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + target_id = vws_client.add_target( + name="example", + width=1, + image=image_file_failed_state, + active_flag=True, + application_metadata=None, + ) + return vws_client.get_target_record(target_id=target_id).status + + # The given processing time of zero seconds means that the target is + # processed immediately. + assert add_target() == TargetStatuses.FAILED + + @staticmethod + def test_targets_persist_between_calls( + high_quality_image: io.BytesIO, + ) -> None: + """A mock instance keeps its targets between calls of a decorated + function. + """ + database = CloudDatabase() + 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, + ) + vws_client.add_target( + name=f"example-{len(vws_client.list_targets())}", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + return vws_client.list_targets() + + 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 + ) + + @staticmethod + def test_query(high_quality_image: io.BytesIO) -> None: + """Query requests are mocked within the decorated function.""" + database = CloudDatabase() + mock = MockVWS(processing_time_seconds=0) + mock.add_cloud_database(cloud_database=database) + + @mock + def query() -> tuple[str, list[str]]: + """Add a target and query for it.""" + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + target_id = vws_client.add_target( + name="example", + width=1, + image=high_quality_image, + active_flag=True, + application_metadata=None, + ) + vws_client.wait_for_target_processed(target_id=target_id) + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + transport=HTTPXTransport(), + ) + matches = cloud_reco_client.query(image=high_quality_image) + return target_id, [match.target_id for match in matches] + + added_target_id, matching_target_ids = query() + assert matching_target_ids == [added_target_id] + + @staticmethod + def test_decorating_a_method() -> None: + """It is possible to decorate a method.""" + database = CloudDatabase() + mock = MockVWS() + mock.add_cloud_database(cloud_database=database) + + class _Example: + """A class with a decorated method.""" + + @mock + def get_database_name(self) -> str: + """Get the name of the database from the mock.""" + assert self is not None + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + ) + return vws_client.get_database_summary_report().name + + assert _Example().get_database_name() == database.database_name