import copy
from typing
import TYPE_CHECKING
from typing
import List
from typing
import Optional
from typing
import Tuple
if TYPE_CHECKING:
# pragma: no cover
# import only for linter run
from requests
import PreparedRequest
from responses
import BaseResponse
class FirstMatchRegistry:
def __init__(self) ->
None:
self._responses: List[
"BaseResponse"] = []
@property
def registered(self) -> List[
"BaseResponse"]:
return self._responses
def reset(self) ->
None:
self._responses = []
def find(
self, request:
"PreparedRequest"
) -> Tuple[Optional[
"BaseResponse"], List[str]]:
found =
None
found_match =
None
match_failed_reasons = []
for i, response
in enumerate(self.registered):
match_result, reason = response.matches(request)
if match_result:
if found
is None:
found = i
found_match = response
else:
if self.registered[found].call_count >
0:
# that assumes that some responses were added between calls
self.registered.pop(found)
found_match = response
break
# Multiple matches found. Remove & return the first response.
return self.registered.pop(found), match_failed_reasons
else:
match_failed_reasons.append(reason)
return found_match, match_failed_reasons
def add(self, response:
"BaseResponse") ->
"BaseResponse":
if any(response
is resp
for resp
in self.registered):
# if user adds multiple responses that reference the same instance.
# do a comparison by memory allocation address.
# see https://github.com/getsentry/responses/issues/479
response = copy.deepcopy(response)
self.registered.append(response)
return response
def remove(self, response:
"BaseResponse") -> List[
"BaseResponse"]:
removed_responses = []
while response
in self.registered:
self.registered.remove(response)
removed_responses.append(response)
return removed_responses
def replace(self, response:
"BaseResponse") ->
"BaseResponse":
try:
index = self.registered.index(response)
except ValueError:
raise ValueError(f
"Response is not registered for URL {response.url}")
self.registered[index] = response
return response
class OrderedRegistry(FirstMatchRegistry):
"""Registry where `Response` objects are dependent on the insertion order and invocation index.
OrderedRegistry applies the rule of first
in - first out. Responses should be invoked
in
the same order
in which they were added to the registry. Otherwise, an error
is returned.
"""
def find(
self, request:
"PreparedRequest"
) -> Tuple[Optional[
"BaseResponse"], List[str]]:
"""Find the next registered `Response` and check if it matches the request.
Search
is performed by taking the first element of the registered responses list
and removing this object (popping
from the list).
Parameters
----------
request : PreparedRequest
Request that was caught by the custom adapter.
Returns
-------
Tuple[Optional[
"BaseResponse"], List[str]]
Matched `Response` object
and empty list
in case of match.
Otherwise,
None and a list
with reasons
for not finding a match.
"""
if not self.registered:
return None, [
"No more registered responses"]
response = self.registered.pop(
0)
match_result, reason = response.matches(request)
if not match_result:
self.reset()
self.add(response)
reason = (
"Next 'Response' in the order doesn't match "
f
"due to the following reason: {reason}."
)
return None, [reason]
return response, []