Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions src/mock_vws/_httpx2_mock_server/decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,10 +285,7 @@ def _mock_routes(
for route in api.routes:
url_pattern = base_url.rstrip("/") + route.path_pattern + "$"
compiled_url_pattern = re.compile(pattern=url_pattern)
handler: _Handler = getattr( # pylint: disable=bad-builtin
api,
route.route_name,
) # ty: ignore[unsound-assignment]
handler = route.handler
httpx2_handler = delayed_httpx2_handler(
handler=_httpx2_handler(handler=handler, base_path=base_path),
delay_seconds=response_delay_seconds,
Expand Down
36 changes: 33 additions & 3 deletions src/mock_vws/_mock_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,9 @@
import email.utils
import json
import uuid
from collections.abc import Iterable, Mapping
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from functools import partial
from typing import Any, Final, override

from beartype import beartype
Expand Down Expand Up @@ -71,22 +72,51 @@ class RequestData:
body: bytes


type RouteHandler = Callable[
[RequestData],
tuple[int, Mapping[str, str], str | bytes],
]


@beartype
@dataclass(frozen=True, kw_only=True)
class Route:
"""A representation of a VWS route.

Args:
route_name: The name of the method.
handler: The callback which handles the route.
path_pattern: The end part of a URL pattern. E.g. `/targets` or
`/targets/.+`.
http_methods: HTTP methods that map to the route function.
"""

route_name: str
handler: RouteHandler
path_pattern: str
http_methods: Iterable[str]


@beartype
@dataclass(frozen=True, kw_only=True)
class RouteDefinition[API]:
"""An API route whose handler has not yet been bound to an
instance.
"""

handler: Callable[
[API, RequestData],
tuple[int, Mapping[str, str], str | bytes],
]
path_pattern: str
http_methods: Iterable[str]

def bind(self, *, api: API) -> Route:
"""Bind the route handler to an API instance."""
return Route(
handler=partial(self.handler, api),
path_pattern=self.path_pattern,
http_methods=self.http_methods,
)


@beartype
def _upload_order_key(target: ImageTarget) -> tuple[datetime.datetime, str]:
Expand Down
34 changes: 12 additions & 22 deletions src/mock_vws/_requests_mock_server/mock_web_query_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,10 @@
import email.utils
from collections.abc import Callable, Iterable, Mapping
from http import HTTPMethod, HTTPStatus
from typing import ParamSpec, Protocol, runtime_checkable

from beartype import beartype

from mock_vws._mock_common import RequestData, Route
from mock_vws._mock_common import RequestData, Route, RouteDefinition
from mock_vws._query_tools import (
get_query_match_response_text,
)
Expand All @@ -23,29 +22,20 @@
from mock_vws.image_matchers import ImageMatcher
from mock_vws.target_manager import TargetManager

_ROUTES: set[Route] = set()

_ResponseType = tuple[int, Mapping[str, str], str | bytes]
_P = ParamSpec("_P")


@runtime_checkable
class _RouteMethod(Protocol[_P]):
"""Callable used for routing which also exposes ``__name__``."""

__name__: str

def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _ResponseType:
"""Return a mock response."""
... # pylint: disable=unnecessary-ellipsis
type _RouteMethod = Callable[
["MockVuforiaWebQueryAPI", RequestData],
_ResponseType,
]
_ROUTES: set[RouteDefinition[MockVuforiaWebQueryAPI]] = set()


@beartype
def route(
*,
path_pattern: str,
http_methods: Iterable[str],
) -> Callable[[_RouteMethod[_P]], _RouteMethod[_P]]:
) -> Callable[[_RouteMethod], _RouteMethod]:
"""Register a decorated method so that it can be recognized as a route.

Args:
Expand All @@ -58,17 +48,17 @@ def route(
"""

def decorator(
method: _RouteMethod[_P],
) -> _RouteMethod[_P]:
method: _RouteMethod,
) -> _RouteMethod:
"""Register a decorated method so that it can be recognized as a
route.

Returns:
The given `method` with multiple changes, including added
validators.
"""
new_route = Route(
route_name=method.__name__,
new_route = RouteDefinition(
handler=method,
path_pattern=path_pattern,
http_methods=frozenset(http_methods),
)
Expand Down Expand Up @@ -101,7 +91,7 @@ def __init__(
Attributes:
routes: The `Route`s to be used in the mock.
"""
self.routes = _ROUTES
self.routes: set[Route] = {route.bind(api=self) for route in _ROUTES}
self._target_manager = target_manager
self._query_match_checker = query_match_checker
self._failure_response = failure_response
Expand Down
34 changes: 13 additions & 21 deletions src/mock_vws/_requests_mock_server/mock_web_services_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
import uuid
from collections.abc import Callable, Iterable, Mapping
from http import HTTPMethod, HTTPStatus
from typing import TYPE_CHECKING, Any, ParamSpec, Protocol, runtime_checkable
from typing import TYPE_CHECKING, Any
from zoneinfo import ZoneInfo

from beartype import BeartypeConf, beartype
Expand All @@ -30,6 +30,7 @@
RECO_COUNTS_REPORT_PATH_PATTERN,
RequestData,
Route,
RouteDefinition,
json_dump,
sorted_targets,
)
Expand Down Expand Up @@ -78,29 +79,20 @@
_MODEL_TARGET_DATASET_UUID_PATTERN = "[A-Za-z0-9-]+"


_ROUTES: set[Route] = set()

_ResponseType = tuple[int, Mapping[str, str], str | bytes]
_P = ParamSpec("_P")


@runtime_checkable
class _RouteMethod(Protocol[_P]):
"""Callable used for routing which also exposes ``__name__``."""

__name__: str

def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _ResponseType:
"""Return a mock response."""
... # pylint: disable=unnecessary-ellipsis
type _RouteMethod = Callable[
["MockVuforiaWebServicesAPI", RequestData],
_ResponseType,
]
_ROUTES: set[RouteDefinition[MockVuforiaWebServicesAPI]] = set()


@beartype
def route(
*,
path_pattern: str,
http_methods: Iterable[HTTPMethod],
) -> Callable[[_RouteMethod[_P]], _RouteMethod[_P]]:
) -> Callable[[_RouteMethod], _RouteMethod]:
"""Register a decorated method so that it can be recognized as a route.

Args:
Expand All @@ -114,17 +106,17 @@ def route(

@beartype
def decorator(
method: _RouteMethod[_P],
) -> _RouteMethod[_P]:
method: _RouteMethod,
) -> _RouteMethod:
"""Register a decorated method so that it can be recognized as a
route.

Returns:
The given `method` with multiple changes, including added
validators.
"""
new_route = Route(
route_name=method.__name__,
new_route = RouteDefinition(
handler=method,
path_pattern=path_pattern,
http_methods=frozenset(http_methods),
)
Expand Down Expand Up @@ -186,7 +178,7 @@ def __init__(
"""
self._target_manager = target_manager
self._base_vws_url = base_vws_url
self.routes = _ROUTES
self.routes: set[Route] = {route.bind(api=self) for route in _ROUTES}
self._processing_time_seconds = processing_time_seconds
self._model_target_generation_failure = model_target_generation_failure
self._model_target_failure_response = model_target_failure_response
Expand Down
5 changes: 1 addition & 4 deletions src/mock_vws/_respx_mock_server/decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -152,10 +152,7 @@ def start_respx_router(
compiled_url_pattern = re.compile(pattern=url_pattern)

for http_method in route.http_methods:
original_callback = getattr( # pylint: disable=bad-builtin
api,
route.route_name,
)
original_callback = route.handler
_ = router.route(
method=http_method,
url=compiled_url_pattern,
Expand Down
10 changes: 5 additions & 5 deletions src/mock_vws/decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,10 @@
Httpx2Router,
start_httpx2_router,
)
from mock_vws._mock_common import MissingSchemeError, RequestData
from mock_vws._mock_common import (
MissingSchemeError,
RequestData,
)
from mock_vws._requests_mock_server.mock_web_query_api import (
MockVuforiaWebQueryAPI,
)
Expand Down Expand Up @@ -513,10 +516,7 @@ def __enter__(self) -> Self:
compiled_url_pattern = re.compile(pattern=url_pattern)

for http_method in route.http_methods:
original_callback = getattr( # pylint: disable=bad-builtin
api,
route.route_name,
)
original_callback = route.handler
_ = mock.add_callback(
method=http_method,
url=compiled_url_pattern,
Expand Down