Merge branch 'switch-to-responses' into more-beartype

Author: adamtheturtle

docs/source/docker.rst

@@ -136,7 +136,7 @@ VWS container
 
    The number of seconds to process each image for.
 
-   Default: ``2``
+   Default: ``2.0``
 
 .. envvar:: DUPLICATES_IMAGE_MATCHER
 

pyproject.toml

@@ -45,7 +45,7 @@ dependencies = [
     "piq",
     "pydantic-settings",
     "requests",
-    "requests-mock",
+    "responses",
     "torch",
     "torchmetrics",
     "tzdata; sys_platform=='win32'",

src/mock_vws/_flask_server/vws.py

@@ -12,7 +12,6 @@
 import uuid
 from enum import StrEnum, auto
 from http import HTTPMethod, HTTPStatus
-from typing import SupportsFloat
 
 import requests
 from beartype import beartype
@@ -68,7 +67,7 @@ class VWSSettings(BaseSettings):
     """Settings for the VWS Flask app."""
 
     target_manager_base_url: str
-    processing_time_seconds: SupportsFloat = 2
+    processing_time_seconds: float = 2.0
     vws_host: str = ""
     duplicates_image_matcher: _ImageMatcherChoice = (
         _ImageMatcherChoice.STRUCTURAL_SIMILARITY

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

@@ -4,11 +4,11 @@
 
 import re
 from contextlib import ContextDecorator
-from typing import Literal, Self, SupportsFloat
+from typing import Literal, Self
 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 (
@@ -39,7 +39,7 @@ def __init__(
         base_vwq_url: str = "https://cloudreco.vuforia.com",
         duplicate_match_checker: ImageMatcher = _STRUCTURAL_SIMILARITY_MATCHER,
         query_match_checker: ImageMatcher = _STRUCTURAL_SIMILARITY_MATCHER,
-        processing_time_seconds: SupportsFloat = 2,
+        processing_time_seconds: float = 2.0,
         target_tracking_rater: TargetTrackingRater = _BRISQUE_TRACKING_RATER,
         *,
         real_http: bool = False,
@@ -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,32 +116,48 @@ def __enter__(self) -> Self:
         Returns:
             ``self``.
         """
-        with Mocker(real_http=self._real_http) as mock:
-            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_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.add_callback(
+                    method=vws_http_method,
+                    url=compiled_url_pattern,
+                    callback=getattr(self._mock_vws_api, vws_route.route_name),
+                    content_type=None,
                 )
 
-                for vws_http_method in vws_route.http_methods:
-                    mock.register_uri(
-                        method=vws_http_method,
-                        url=re.compile(url_pattern),
-                        text=getattr(self._mock_vws_api, vws_route.route_name),
-                    )
-
-            for vwq_route in self._mock_vwq_api.routes:
-                url_pattern = urljoin(
-                    base=self._base_vwq_url,
-                    url=f"{vwq_route.path_pattern}$",
+        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.add_callback(
+                    method=vwq_http_method,
+                    url=compiled_url_pattern,
+                    callback=getattr(self._mock_vwq_api, vwq_route.route_name),
+                    content_type=None,
                 )
 
-                for vwq_http_method in vwq_route.http_methods:
-                    mock.register_uri(
-                        method=vwq_http_method,
-                        url=re.compile(url_pattern),
-                        text=getattr(self._mock_vwq_api, vwq_route.route_name),
-                    )
+        if self._real_http:
+            combined_pattern = "|".join(
+                f"(?:{pattern.pattern})" for pattern in compiled_url_patterns
+            )
+            negated_pattern = f"(?!{combined_pattern})."
+            compiled_negated_pattern = re.compile(pattern=negated_pattern)
+            mock.add_passthru(prefix=compiled_negated_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,18 +22,15 @@
 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 = tuple[int, dict[str, str], str]
+
 
 def route(
     path_pattern: str,
     http_methods: set[str],
-) -> Callable[[Callable[..., str]], Callable[..., str]]:
+) -> Callable[[Callable[..., _ResponseType]], Callable[..., _ResponseType]]:
     """
     Register a decorated method so that it can be recognized as a route.
 
@@ -45,7 +43,9 @@ def route(
         A decorator which takes methods and makes them recognizable as routes.
     """
 
-    def decorator(method: Callable[..., str]) -> Callable[..., str]:
+    def decorator(
+        method: Callable[..., _ResponseType],
+    ) -> Callable[..., _ResponseType]:
         """
         Register a decorated method so that it can be recognized as a route.
 
@@ -66,18 +66,22 @@ def decorator(method: Callable[..., str]) -> Callable[..., str]:
     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__(
@@ -99,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") -> str:
+    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,
         )
@@ -130,11 +132,11 @@ def query(self, request: "Request", context: "Context") -> str:
             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