Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix interoperability with responses 0.17.0 (#4749)

Browse Source
This commit is contained in:
Bert Blommers 2022-01-12 12:16:02 -01:00 committed by GitHub
parent a688c00325
commit 2407e969ac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 220 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.github/workflows/build.yml vendored
View File

@ -217,12 +217,13 @@ jobs:
serverlogs/*
test_responses:
name: Test Responses==0.12.0
name: Test Responses versions
runs-on: ubuntu-latest
needs: lint
strategy:
matrix:
python-version: [ 3.7, 3.8 ]
python-version: [ 3.8 ]
responses-version: [0.11.0, 0.12.0, 0.12.1, 0.13.0, 0.15.0, 0.17.0]
steps:
- uses: actions/checkout@v2
@ -246,9 +247,9 @@ jobs:
python -m pip install --upgrade pip
pip install -r requirements-dev.txt
pip install pytest-cov
pip install responses==0.12.0
pip install responses==${{ matrix.responses-version }}
pip install "coverage<=4.5.4"
- name: Test core-logic with responses==0.12.0
- name: Test core-logic with responses==${{ matrix.responses-version }}
run: |
pytest -sv --cov=moto --cov-report xml ./tests/test_core
- name: "Upload coverage to Codecov"

2
codecov.yml
View File

@ -1,7 +1,7 @@
codecov:
notify:
# Leave a GitHub comment after all builds have passed
after_n_builds: 10
after_n_builds: 14
coverage:
status:
project:

2
moto/apigateway/models.py
View File

@ -920,7 +920,7 @@ class RestAPI(CloudFormationModel):
content_type="text/plain",
match_querystring=False,
)
responses_mock._matches.insert(0, callback_response)
responses_mock.add(callback_response)
def create_authorizer(
self,

171
moto/core/custom_responses_mock.py Normal file
View File

@ -0,0 +1,171 @@
import responses
import types
from io import BytesIO
from http.client import responses as http_responses
from urllib.parse import urlparse
from werkzeug.wrappers import Request
from moto.utilities.distutils_version import LooseVersion
try:
from importlib.metadata import version
except ImportError:
from importlib_metadata import version
class CallbackResponse(responses.CallbackResponse):
"""
Need to subclass so we can change a couple things
"""
def get_response(self, request):
"""
Need to override this so we can pass decode_content=False
"""
if not isinstance(request, Request):
url = urlparse(request.url)
if request.body is None:
body = None
elif isinstance(request.body, str):
body = BytesIO(request.body.encode("UTF-8"))
elif hasattr(request.body, "read"):
body = BytesIO(request.body.read())
else:
body = BytesIO(request.body)
req = Request.from_values(
path="?".join([url.path, url.query]),
input_stream=body,
content_length=request.headers.get("Content-Length"),
content_type=request.headers.get("Content-Type"),
method=request.method,
base_url="{scheme}://{netloc}".format(
scheme=url.scheme, netloc=url.netloc
),
headers=[(k, v) for k, v in request.headers.items()],
)
request = req
headers = self.get_headers()
result = self.callback(request)
if isinstance(result, Exception):
raise result
status, r_headers, body = result
body = responses._handle_body(body)
headers.update(r_headers)
return responses.HTTPResponse(
status=status,
reason=http_responses.get(status),
body=body,
headers=headers,
preload_content=False,
# Need to not decode_content to mimic requests
decode_content=False,
)
def _url_matches(self, url, other, match_querystring=False):
"""
Need to override this so we can fix querystrings breaking regex matching
"""
if not match_querystring:
other = other.split("?", 1)[0]
if responses._is_string(url):
if responses._has_unicode(url):
url = responses._clean_unicode(url)
if not isinstance(other, str):
other = other.encode("ascii").decode("utf8")
return self._url_matches_strict(url, other)
elif isinstance(url, responses.Pattern) and url.match(other):
return True
else:
return False
def not_implemented_callback(request):
status = 400
headers = {}
response = "The method is not implemented"
return status, headers, response
# Modify behaviour of the matcher to only/always return the first match
# Default behaviour is to return subsequent matches for subsequent requests, which leads to https://github.com/spulec/moto/issues/2567
# - First request matches on the appropriate S3 URL
# - Same request, executed again, will be matched on the subsequent match, which happens to be the catch-all, not-yet-implemented, callback
# Fix: Always return the first match
def _find_first_match_legacy(self, request):
matches = [match for match in self._matches if match.matches(request)]
# Look for implemented callbacks first
implemented_matches = [
m
for m in matches
if type(m) is not CallbackResponse or m.callback != not_implemented_callback
]
if implemented_matches:
return implemented_matches[0]
elif matches:
# We had matches, but all were of type not_implemented_callback
return matches[0]
return None
# Internal API changed: this method should be used for Responses 0.12.1 < .. < 0.17.0
def _find_first_match(self, request):
matches = []
match_failed_reasons = []
for match in self._matches:
match_result, reason = match.matches(request)
if match_result:
matches.append(match)
else:
match_failed_reasons.append(reason)
# Look for implemented callbacks first
implemented_matches = [
m
for m in matches
if type(m) is not CallbackResponse or m.callback != not_implemented_callback
]
if implemented_matches:
return implemented_matches[0], []
elif matches:
# We had matches, but all were of type not_implemented_callback
return matches[0], match_failed_reasons
return None, match_failed_reasons
def get_response_mock():
"""
The responses-library is crucial in ensuring that requests to AWS are intercepted, and routed to the right backend.
However, as our usecase is different from a 'typical' responses-user, Moto always needs some custom logic to ensure responses behaves in a way that works for us.
For older versions, that meant changing the internal logic
For later versions, > 0.17.0, we can use a custom registry, and extend the logic instead of overriding it
For all versions, we need to add passthrough to allow non-AWS requests to work
"""
responses_mock = None
RESPONSES_VERSION = version("responses")
if LooseVersion(RESPONSES_VERSION) < LooseVersion("0.12.1"):
responses_mock = responses.RequestsMock(assert_all_requests_are_fired=False)
responses_mock._find_match = types.MethodType(
_find_first_match_legacy, responses_mock
)
elif LooseVersion(RESPONSES_VERSION) >= LooseVersion("0.17.0"):
from .responses_custom_registry import CustomRegistry
responses_mock = responses.RequestsMock(
assert_all_requests_are_fired=False, registry=CustomRegistry
)
else:
responses_mock = responses.RequestsMock(assert_all_requests_are_fired=False)
responses_mock._find_match = types.MethodType(_find_first_match, responses_mock)
responses_mock.add_passthru("http")
return responses_mock

156
moto/core/models.py
View File

@ -6,28 +6,23 @@ import os
import random
import re
import string
import types
from abc import abstractmethod
from io import BytesIO
from collections import defaultdict
try:
from importlib.metadata import version
except ImportError:
from importlib_metadata import version
from botocore.config import Config
from botocore.handlers import BUILTIN_HANDLERS
from botocore.awsrequest import AWSResponse
from http.client import responses as http_responses
from urllib.parse import urlparse
from werkzeug.wrappers import Request
from moto import settings
import responses
from moto.packages.httpretty import HTTPretty
from moto.utilities.distutils_version import LooseVersion
from unittest.mock import patch
from .custom_responses_mock import (
get_response_mock,
CallbackResponse,
not_implemented_callback,
)
from .utils import (
convert_httpretty_response,
convert_regex_to_flask_path,
@ -214,141 +209,12 @@ RESPONSES_METHODS = [
]
class CallbackResponse(responses.CallbackResponse):
"""
Need to subclass so we can change a couple things
"""
def get_response(self, request):
"""
Need to override this so we can pass decode_content=False
"""
if not isinstance(request, Request):
url = urlparse(request.url)
if request.body is None:
body = None
elif isinstance(request.body, str):
body = BytesIO(request.body.encode("UTF-8"))
elif hasattr(request.body, "read"):
body = BytesIO(request.body.read())
else:
body = BytesIO(request.body)
req = Request.from_values(
path="?".join([url.path, url.query]),
input_stream=body,
content_length=request.headers.get("Content-Length"),
content_type=request.headers.get("Content-Type"),
method=request.method,
base_url="{scheme}://{netloc}".format(
scheme=url.scheme, netloc=url.netloc
),
headers=[(k, v) for k, v in request.headers.items()],
)
request = req
headers = self.get_headers()
result = self.callback(request)
if isinstance(result, Exception):
raise result
status, r_headers, body = result
body = responses._handle_body(body)
headers.update(r_headers)
return responses.HTTPResponse(
status=status,
reason=http_responses.get(status),
body=body,
headers=headers,
preload_content=False,
# Need to not decode_content to mimic requests
decode_content=False,
)
def _url_matches(self, url, other, match_querystring=False):
"""
Need to override this so we can fix querystrings breaking regex matching
"""
if not match_querystring:
other = other.split("?", 1)[0]
if responses._is_string(url):
if responses._has_unicode(url):
url = responses._clean_unicode(url)
if not isinstance(other, str):
other = other.encode("ascii").decode("utf8")
return self._url_matches_strict(url, other)
elif isinstance(url, responses.Pattern) and url.match(other):
return True
else:
return False
botocore_mock = responses.RequestsMock(
assert_all_requests_are_fired=False,
target="botocore.vendored.requests.adapters.HTTPAdapter.send",
)
responses_mock = responses.RequestsMock(assert_all_requests_are_fired=False)
# Add passthrough to allow any other requests to work
# Since this uses .startswith, it applies to http and https requests.
responses_mock.add_passthru("http")
def _find_first_match_legacy(self, request):
matches = [match for match in self._matches if match.matches(request)]
# Look for implemented callbacks first
implemented_matches = [
m
for m in matches
if type(m) is not CallbackResponse or m.callback != not_implemented_callback
]
if implemented_matches:
return implemented_matches[0]
elif matches:
# We had matches, but all were of type not_implemented_callback
return matches[0]
return None
def _find_first_match(self, request):
matches = []
match_failed_reasons = []
for match in self._matches:
match_result, reason = match.matches(request)
if match_result:
matches.append(match)
else:
match_failed_reasons.append(reason)
# Look for implemented callbacks first
implemented_matches = [
m
for m in matches
if type(m) is not CallbackResponse or m.callback != not_implemented_callback
]
if implemented_matches:
return implemented_matches[0], []
elif matches:
# We had matches, but all were of type not_implemented_callback
return matches[0], match_failed_reasons
return None, match_failed_reasons
# Modify behaviour of the matcher to only/always return the first match
# Default behaviour is to return subsequent matches for subsequent requests, which leads to https://github.com/spulec/moto/issues/2567
# - First request matches on the appropriate S3 URL
# - Same request, executed again, will be matched on the subsequent match, which happens to be the catch-all, not-yet-implemented, callback
# Fix: Always return the first match
RESPONSES_VERSION = version("responses")
if LooseVersion(RESPONSES_VERSION) < LooseVersion("0.12.1"):
responses_mock._find_match = types.MethodType(
_find_first_match_legacy, responses_mock
)
else:
responses_mock._find_match = types.MethodType(_find_first_match, responses_mock)
responses_mock = get_response_mock()
BOTOCORE_HTTP_METHODS = ["GET", "DELETE", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
@ -448,18 +314,10 @@ def patch_resource(resource):
raise Exception(f"Argument {resource} should be of type boto3.resource")
def not_implemented_callback(request):
status = 400
headers = {}
response = "The method is not implemented"
return status, headers, response
class BotocoreEventMockAWS(BaseMockAWS):
def reset(self):
botocore_stubber.reset()
responses_mock.reset()
responses_mock.calls.reset()
def enable_patching(self):
botocore_stubber.enabled = True

35
moto/core/responses_custom_registry.py Normal file
View File

@ -0,0 +1,35 @@
# This will only exist in responses >= 0.17
from responses import registries
from .custom_responses_mock import CallbackResponse, not_implemented_callback
class CustomRegistry(registries.FirstMatchRegistry):
"""
Custom Registry that returns requests in an order that makes sense for Moto:
- Implemented callbacks take precedence over non-implemented-callbacks
- CallbackResponses are not discarded after first use - users can mock the same URL as often as they like
"""
def find(self, request):
found = []
match_failed_reasons = []
for response in self.registered:
match_result, reason = response.matches(request)
if match_result:
found.append(response)
else:
match_failed_reasons.append(reason)
# Look for implemented callbacks first
implemented_matches = [
m
for m in found
if type(m) is not CallbackResponse or m.callback != not_implemented_callback
]
if implemented_matches:
return implemented_matches[0], match_failed_reasons
elif found:
# We had matches, but all were of type not_implemented_callback
return found[0], match_failed_reasons
else:
return None, match_failed_reasons