Merge pull request #2264 from VWS-Python/switch-to-responses

Switch to responses from requests_mock

Author: adamtheturtle

pyproject.toml

@@ -44,7 +44,7 @@ dependencies = [
     "piq",
     "pydantic-settings",
     "requests",
-    "requests-mock",
+    "responses",
     "torch",
     "torchmetrics",
     "tzdata; sys_platform=='win32'",
@@ -78,7 +78,7 @@ optional-dependencies.dev = [
     "pytest-xdist==3.6.1",
     "python-dotenv==1.0.1",
     "pyyaml==6.0.2",
-    "requests-mock-flask==2023.5.14",
+    "requests-mock-flask==2024.8.30.1",
     "ruff==0.6.2",
     "sphinx==8.0.2",
     "sphinx-copybutton==0.5.2",

src/mock_vws/_requests_mock_server/__init__.py

@@ -1,3 +1,3 @@
 """
-An interface to the mock Vuforia which uses ``requests_mock``.
+An interface to the mock Vuforia which uses ``responses``.
 """

src/mock_vws/_requests_mock_server/decorators.py

@@ -8,7 +8,7 @@
 from urllib.parse import urljoin, urlparse
 
 import requests
-from requests_mock.mocker import Mocker
+from responses import RequestsMock
 
 from mock_vws.database import VuforiaDatabase
 from mock_vws.image_matchers import (
@@ -69,7 +69,7 @@ def __init__(
         """
         super().__init__()
         self._real_http = real_http
-        self._mock: Mocker
+        self._mock: RequestsMock
         self._target_manager = TargetManager()
 
         self._base_vws_url = base_vws_url
@@ -116,33 +116,45 @@ def __enter__(self) -> Self:
         Returns:
             ``self``.
         """
-        mock = Mocker(real_http=self._real_http)
+        compiled_url_patterns: set[re.Pattern[str]] = set()
+
+        mock = RequestsMock(assert_all_requests_are_fired=False)
         for vws_route in self._mock_vws_api.routes:
             url_pattern = urljoin(
                 base=self._base_vws_url,
                 url=f"{vws_route.path_pattern}$",
             )
+            compiled_url_pattern = re.compile(pattern=url_pattern)
+            compiled_url_patterns.add(compiled_url_pattern)
 
             for vws_http_method in vws_route.http_methods:
-                mock.register_uri(
+                mock.add_callback(
                     method=vws_http_method,
-                    url=re.compile(url_pattern),
-                    text=getattr(self._mock_vws_api, vws_route.route_name),
+                    url=compiled_url_pattern,
+                    callback=getattr(self._mock_vws_api, vws_route.route_name),
+                    content_type=None,
                 )
 
         for vwq_route in self._mock_vwq_api.routes:
             url_pattern = urljoin(
                 base=self._base_vwq_url,
                 url=f"{vwq_route.path_pattern}$",
             )
+            compiled_url_pattern = re.compile(pattern=url_pattern)
+            compiled_url_patterns.add(compiled_url_pattern)
 
             for vwq_http_method in vwq_route.http_methods:
-                mock.register_uri(
+                mock.add_callback(
                     method=vwq_http_method,
-                    url=re.compile(url_pattern),
-                    text=getattr(self._mock_vwq_api, vwq_route.route_name),
+                    url=compiled_url_pattern,
+                    callback=getattr(self._mock_vwq_api, vwq_route.route_name),
+                    content_type=None,
                 )
 
+        if self._real_http:
+            all_requests_pattern = re.compile(pattern=".*")
+            mock.add_passthru(prefix=all_requests_pattern)
+
         self._mock = mock
         self._mock.start()
 

src/mock_vws/_requests_mock_server/mock_web_query_api.py

@@ -7,8 +7,9 @@
 
 import email.utils
 from collections.abc import Callable
-from http import HTTPMethod
-from typing import TYPE_CHECKING
+from http import HTTPMethod, HTTPStatus
+
+from requests.models import PreparedRequest
 
 from mock_vws._mock_common import Route
 from mock_vws._query_tools import (
@@ -21,14 +22,9 @@
 from mock_vws.image_matchers import ImageMatcher
 from mock_vws.target_manager import TargetManager
 
-if TYPE_CHECKING:
-    from requests_mock.request import Request
-    from requests_mock.response import Context
-
-
 _ROUTES: set[Route] = set()
 
-_ResponseType = str
+_ResponseType = tuple[int, dict[str, str], str]
 
 
 def route(
@@ -70,18 +66,22 @@ def decorator(
     return decorator
 
 
-def _body_bytes(request: "Request") -> bytes:
+def _body_bytes(request: PreparedRequest) -> bytes:
     """
     Return the body of a request as bytes.
     """
-    return request.body or b""
+    if request.body is None:
+        return b""
+
+    assert isinstance(request.body, bytes)
+    return request.body
 
 
 class MockVuforiaWebQueryAPI:
     """
     A fake implementation of the Vuforia Web Query API.
 
-    This implementation is tied to the implementation of `requests_mock`.
+    This implementation is tied to the implementation of ``responses``.
     """
 
     def __init__(
@@ -103,28 +103,26 @@ def __init__(
         self._query_match_checker = query_match_checker
 
     @route(path_pattern="/v1/query", http_methods={HTTPMethod.POST})
-    def query(self, request: "Request", context: "Context") -> _ResponseType:
+    def query(self, request: PreparedRequest) -> _ResponseType:
         """
         Perform an image recognition query.
         """
         try:
             run_query_validators(
-                request_path=request.path,
+                request_path=request.path_url,
                 request_headers=request.headers,
                 request_body=_body_bytes(request=request),
-                request_method=request.method,
+                request_method=request.method or "",
                 databases=self._target_manager.databases,
             )
         except ValidatorError as exc:
-            context.headers = exc.headers
-            context.status_code = exc.status_code
-            return exc.response_text
+            return exc.status_code, exc.headers, exc.response_text
 
         response_text = get_query_match_response_text(
             request_headers=request.headers,
             request_body=_body_bytes(request=request),
-            request_method=request.method,
-            request_path=request.path,
+            request_method=request.method or "",
+            request_path=request.path_url,
             databases=self._target_manager.databases,
             query_match_checker=self._query_match_checker,
         )
@@ -134,11 +132,11 @@ def query(self, request: "Request", context: "Context") -> _ResponseType:
             localtime=False,
             usegmt=True,
         )
-        context.headers = {
+        headers = {
             "Connection": "keep-alive",
             "Content-Type": "application/json",
             "Server": "nginx",
             "Date": date,
             "Content-Length": str(len(response_text)),
         }
-        return response_text
+        return HTTPStatus.OK, headers, response_text