Changeset View
Changeset View
Standalone View
Standalone View
swh/core/pytest_plugin.py
Show All 21 Lines | |||||
# Check get_local_factory function | # Check get_local_factory function | ||||
# Maximum number of iteration checks to generate requests responses | # Maximum number of iteration checks to generate requests responses | ||||
MAX_VISIT_FILES = 10 | MAX_VISIT_FILES = 10 | ||||
def get_response_cb( | def get_response_cb( | ||||
request: requests.Request, context, datadir, | request: requests.Request, | ||||
context, | |||||
datadir, | |||||
ignore_urls: List[str] = [], | ignore_urls: List[str] = [], | ||||
visits: Optional[Dict] = None): | visits: Optional[Dict] = None, | ||||
): | |||||
"""Mount point callback to fetch on disk the request's content. The request | """Mount point callback to fetch on disk the request's content. The request | ||||
urls provided are url decoded first to resolve the associated file on disk. | urls provided are url decoded first to resolve the associated file on disk. | ||||
This is meant to be used as 'body' argument of the requests_mock.get() | This is meant to be used as 'body' argument of the requests_mock.get() | ||||
method. | method. | ||||
It will look for files on the local filesystem based on the requested URL, | It will look for files on the local filesystem based on the requested URL, | ||||
using the following rules: | using the following rules: | ||||
Show All 34 Lines | Args: | ||||
file exists | file exists | ||||
visits: Dict of url, number of visits. If None, disable multi visit | visits: Dict of url, number of visits. If None, disable multi visit | ||||
support (default) | support (default) | ||||
Returns: | Returns: | ||||
Optional[FileDescriptor] on disk file to read from the test context | Optional[FileDescriptor] on disk file to read from the test context | ||||
""" | """ | ||||
logger.debug('get_response_cb(%s, %s)', request, context) | logger.debug("get_response_cb(%s, %s)", request, context) | ||||
logger.debug('url: %s', request.url) | logger.debug("url: %s", request.url) | ||||
logger.debug('ignore_urls: %s', ignore_urls) | logger.debug("ignore_urls: %s", ignore_urls) | ||||
unquoted_url = unquote(request.url) | unquoted_url = unquote(request.url) | ||||
if unquoted_url in ignore_urls: | if unquoted_url in ignore_urls: | ||||
context.status_code = 404 | context.status_code = 404 | ||||
return None | return None | ||||
url = urlparse(unquoted_url) | url = urlparse(unquoted_url) | ||||
# http://pypi.org ~> http_pypi.org | # http://pypi.org ~> http_pypi.org | ||||
# https://files.pythonhosted.org ~> https_files.pythonhosted.org | # https://files.pythonhosted.org ~> https_files.pythonhosted.org | ||||
dirname = '%s_%s' % (url.scheme, url.hostname) | dirname = "%s_%s" % (url.scheme, url.hostname) | ||||
# url.path: pypi/<project>/json -> local file: pypi_<project>_json | # url.path: pypi/<project>/json -> local file: pypi_<project>_json | ||||
filename = url.path[1:] | filename = url.path[1:] | ||||
if filename.endswith('/'): | if filename.endswith("/"): | ||||
filename = filename[:-1] | filename = filename[:-1] | ||||
filename = filename.replace('/', '_') | filename = filename.replace("/", "_") | ||||
if url.query: | if url.query: | ||||
filename += ',' + url.query.replace('&', ',') | filename += "," + url.query.replace("&", ",") | ||||
filepath = path.join(datadir, dirname, filename) | filepath = path.join(datadir, dirname, filename) | ||||
if visits is not None: | if visits is not None: | ||||
visit = visits.get(url, 0) | visit = visits.get(url, 0) | ||||
visits[url] = visit + 1 | visits[url] = visit + 1 | ||||
if visit: | if visit: | ||||
filepath = filepath + '_visit%s' % visit | filepath = filepath + "_visit%s" % visit | ||||
if not path.isfile(filepath): | if not path.isfile(filepath): | ||||
logger.debug('not found filepath: %s', filepath) | logger.debug("not found filepath: %s", filepath) | ||||
context.status_code = 404 | context.status_code = 404 | ||||
return None | return None | ||||
fd = open(filepath, 'rb') | fd = open(filepath, "rb") | ||||
context.headers['content-length'] = str(path.getsize(filepath)) | context.headers["content-length"] = str(path.getsize(filepath)) | ||||
return fd | return fd | ||||
@pytest.fixture | @pytest.fixture | ||||
def datadir(request): | def datadir(request): | ||||
"""By default, returns the test directory's data directory. | """By default, returns the test directory's data directory. | ||||
This can be overridden on a per file tree basis. Add an override | This can be overridden on a per file tree basis. Add an override | ||||
definition in the local conftest, for example:: | definition in the local conftest, for example:: | ||||
import pytest | import pytest | ||||
from os import path | from os import path | ||||
@pytest.fixture | @pytest.fixture | ||||
def datadir(): | def datadir(): | ||||
return path.join(path.abspath(path.dirname(__file__)), 'resources') | return path.join(path.abspath(path.dirname(__file__)), 'resources') | ||||
""" | """ | ||||
return path.join(path.dirname(str(request.fspath)), 'data') | return path.join(path.dirname(str(request.fspath)), "data") | ||||
def requests_mock_datadir_factory(ignore_urls: List[str] = [], | def requests_mock_datadir_factory( | ||||
has_multi_visit: bool = False): | ignore_urls: List[str] = [], has_multi_visit: bool = False | ||||
): | |||||
"""This factory generates fixture which allow to look for files on the | """This factory generates fixture which allow to look for files on the | ||||
local filesystem based on the requested URL, using the following rules: | local filesystem based on the requested URL, using the following rules: | ||||
- files are searched in the datadir/<hostname> directory | - files are searched in the datadir/<hostname> directory | ||||
- the local file name is the path part of the URL with path hierarchy | - the local file name is the path part of the URL with path hierarchy | ||||
markers (aka '/') replaced by '_' | markers (aka '/') replaced by '_' | ||||
Show All 14 Lines | - requests_mock_datadir_factory(ignore_urls=['url1', 'url2']): | ||||
returning 404. | returning 404. | ||||
Args: | Args: | ||||
ignore_urls: List of urls to always returns 404 (whether file | ignore_urls: List of urls to always returns 404 (whether file | ||||
exists or not) | exists or not) | ||||
has_multi_visit: Activate or not the multiple visits behavior | has_multi_visit: Activate or not the multiple visits behavior | ||||
""" | """ | ||||
@pytest.fixture | @pytest.fixture | ||||
def requests_mock_datadir(requests_mock, datadir): | def requests_mock_datadir(requests_mock, datadir): | ||||
if not has_multi_visit: | if not has_multi_visit: | ||||
cb = partial(get_response_cb, | cb = partial(get_response_cb, ignore_urls=ignore_urls, datadir=datadir) | ||||
ignore_urls=ignore_urls, | requests_mock.get(re.compile("https?://"), body=cb) | ||||
datadir=datadir) | |||||
requests_mock.get(re.compile('https?://'), body=cb) | |||||
else: | else: | ||||
visits = {} | visits = {} | ||||
requests_mock.get(re.compile('https?://'), body=partial( | requests_mock.get( | ||||
get_response_cb, ignore_urls=ignore_urls, visits=visits, | re.compile("https?://"), | ||||
datadir=datadir) | body=partial( | ||||
get_response_cb, | |||||
ignore_urls=ignore_urls, | |||||
visits=visits, | |||||
datadir=datadir, | |||||
), | |||||
) | ) | ||||
return requests_mock | return requests_mock | ||||
return requests_mock_datadir | return requests_mock_datadir | ||||
# Default `requests_mock_datadir` implementation | # Default `requests_mock_datadir` implementation | ||||
requests_mock_datadir = requests_mock_datadir_factory([]) | requests_mock_datadir = requests_mock_datadir_factory([]) | ||||
# Implementation for multiple visits behavior: | # Implementation for multiple visits behavior: | ||||
# - first time, it checks for a file named `filename` | # - first time, it checks for a file named `filename` | ||||
# - second time, it checks for a file named `filename`_visit1 | # - second time, it checks for a file named `filename`_visit1 | ||||
# etc... | # etc... | ||||
requests_mock_datadir_visits = requests_mock_datadir_factory( | requests_mock_datadir_visits = requests_mock_datadir_factory(has_multi_visit=True) | ||||
has_multi_visit=True) | |||||
@pytest.fixture | @pytest.fixture | ||||
def swh_rpc_client(swh_rpc_client_class, swh_rpc_adapter): | def swh_rpc_client(swh_rpc_client_class, swh_rpc_adapter): | ||||
"""This fixture generates an RPCClient instance that uses the class generated | """This fixture generates an RPCClient instance that uses the class generated | ||||
by the rpc_client_class fixture as backend. | by the rpc_client_class fixture as backend. | ||||
Since it uses the swh_rpc_adapter, HTTP queries will be intercepted and | Since it uses the swh_rpc_adapter, HTTP queries will be intercepted and | ||||
routed directly to the current Flask app (as provided by the `app` | routed directly to the current Flask app (as provided by the `app` | ||||
fixture). | fixture). | ||||
So this stack of fixtures allows to test the RPCClient -> RPCServerApp | So this stack of fixtures allows to test the RPCClient -> RPCServerApp | ||||
communication path using a real RPCClient instance and a real Flask | communication path using a real RPCClient instance and a real Flask | ||||
(RPCServerApp) app instance. | (RPCServerApp) app instance. | ||||
To use this fixture: | To use this fixture: | ||||
- ensure an `app` fixture exists and generate a Flask application, | - ensure an `app` fixture exists and generate a Flask application, | ||||
- implement an `swh_rpc_client_class` fixtures that returns the | - implement an `swh_rpc_client_class` fixtures that returns the | ||||
RPCClient-based class to use as client side for the tests, | RPCClient-based class to use as client side for the tests, | ||||
- implement your tests using this `swh_rpc_client` fixture. | - implement your tests using this `swh_rpc_client` fixture. | ||||
See swh/core/api/tests/test_rpc_client_server.py for an example of usage. | See swh/core/api/tests/test_rpc_client_server.py for an example of usage. | ||||
""" | """ | ||||
url = 'mock://example.com' | url = "mock://example.com" | ||||
cli = swh_rpc_client_class(url=url) | cli = swh_rpc_client_class(url=url) | ||||
# we need to clear the list of existing adapters here so we ensure we | # we need to clear the list of existing adapters here so we ensure we | ||||
# have one and only one adapter which is then used for all the requests. | # have one and only one adapter which is then used for all the requests. | ||||
cli.session.adapters.clear() | cli.session.adapters.clear() | ||||
cli.session.mount('mock://', swh_rpc_adapter) | cli.session.mount("mock://", swh_rpc_adapter) | ||||
return cli | return cli | ||||
@pytest.fixture | @pytest.fixture | ||||
def swh_rpc_adapter(app): | def swh_rpc_adapter(app): | ||||
"""Fixture that generates a requests.Adapter instance that | """Fixture that generates a requests.Adapter instance that | ||||
can be used to test client/servers code based on swh.core.api classes. | can be used to test client/servers code based on swh.core.api classes. | ||||
Show All 10 Lines | class RPCTestAdapter(BaseAdapter): | ||||
def build_response(self, req, resp): | def build_response(self, req, resp): | ||||
response = requests.Response() | response = requests.Response() | ||||
# Fallback to None if there's no status_code, for whatever reason. | # Fallback to None if there's no status_code, for whatever reason. | ||||
response.status_code = resp.status_code | response.status_code = resp.status_code | ||||
# Make headers case-insensitive. | # Make headers case-insensitive. | ||||
response.headers = CaseInsensitiveDict(getattr(resp, 'headers', {})) | response.headers = CaseInsensitiveDict(getattr(resp, "headers", {})) | ||||
# Set encoding. | # Set encoding. | ||||
response.encoding = get_encoding_from_headers(response.headers) | response.encoding = get_encoding_from_headers(response.headers) | ||||
response.raw = resp | response.raw = resp | ||||
response.reason = response.raw.status | response.reason = response.raw.status | ||||
if isinstance(req.url, bytes): | if isinstance(req.url, bytes): | ||||
response.url = req.url.decode('utf-8') | response.url = req.url.decode("utf-8") | ||||
else: | else: | ||||
response.url = req.url | response.url = req.url | ||||
# Give the Response some context. | # Give the Response some context. | ||||
response.request = req | response.request = req | ||||
response.connection = self | response.connection = self | ||||
response._content = resp.data | response._content = resp.data | ||||
return response | return response | ||||
def send(self, request, **kw): | def send(self, request, **kw): | ||||
""" | """ | ||||
Overrides ``requests.adapters.BaseAdapter.send`` | Overrides ``requests.adapters.BaseAdapter.send`` | ||||
""" | """ | ||||
resp = self._client.open( | resp = self._client.open( | ||||
request.url, method=request.method, | request.url, | ||||
method=request.method, | |||||
headers=request.headers.items(), | headers=request.headers.items(), | ||||
data=request.body, | data=request.body, | ||||
) | ) | ||||
return self.build_response(request, resp) | return self.build_response(request, resp) | ||||
@pytest.fixture | @pytest.fixture | ||||
def flask_app_client(app): | def flask_app_client(app): | ||||
with app.test_client() as client: | with app.test_client() as client: | ||||
yield client | yield client | ||||
# stolen from pytest-flask, required to have url_for() working within tests | # stolen from pytest-flask, required to have url_for() working within tests | ||||
# using flask_app_client fixture. | # using flask_app_client fixture. | ||||
@pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
def _push_request_context(request): | def _push_request_context(request): | ||||
"""During tests execution request context has been pushed, e.g. `url_for`, | """During tests execution request context has been pushed, e.g. `url_for`, | ||||
`session`, etc. can be used in tests as is:: | `session`, etc. can be used in tests as is:: | ||||
def test_app(app, client): | def test_app(app, client): | ||||
assert client.get(url_for('myview')).status_code == 200 | assert client.get(url_for('myview')).status_code == 200 | ||||
""" | """ | ||||
if 'app' not in request.fixturenames: | if "app" not in request.fixturenames: | ||||
return | return | ||||
app = request.getfixturevalue('app') | app = request.getfixturevalue("app") | ||||
ctx = app.test_request_context() | ctx = app.test_request_context() | ||||
ctx.push() | ctx.push() | ||||
def teardown(): | def teardown(): | ||||
ctx.pop() | ctx.pop() | ||||
request.addfinalizer(teardown) | request.addfinalizer(teardown) |