Changeset View
Changeset View
Standalone View
Standalone View
swh/core/pytest_plugin.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import logging | import logging | ||||
| import re | import re | ||||
| import pytest | import pytest | ||||
| import requests | import requests | ||||
| from functools import partial | from functools import partial | ||||
| from os import path | from os import path | ||||
| from typing import Dict, List, Optional | from typing import Dict, List, Optional | ||||
| from urllib.parse import urlparse | from urllib.parse import urlparse, unquote | ||||
| from requests.adapters import BaseAdapter | from requests.adapters import BaseAdapter | ||||
| from requests.structures import CaseInsensitiveDict | from requests.structures import CaseInsensitiveDict | ||||
| from requests.utils import get_encoding_from_headers | from requests.utils import get_encoding_from_headers | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| # 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(request, context, datadir, | def get_response_cb( | ||||
| 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. | """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. | |||||
| 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: | ||||
| - files are searched in the datadir/<hostname> directory | - files are searched in the datadir/<hostname> directory | ||||
| Show All 19 Lines | or a call requests.get like: | ||||
| requests.get('http://nowhere.com/path/to/resource?a=b&c=d') | requests.get('http://nowhere.com/path/to/resource?a=b&c=d') | ||||
| will look the content of the response in: | will look the content of the response in: | ||||
| datadir/http_nowhere.com/path_to_resource,a=b,c=d | datadir/http_nowhere.com/path_to_resource,a=b,c=d | ||||
| Args: | Args: | ||||
| request (requests.Request): Object requests | request: Object requests | ||||
| context (requests.Context): Object holding response metadata | context (requests.Context): Object holding response metadata | ||||
| information (status_code, headers, etc...) | information (status_code, headers, etc...) | ||||
| datadir: Data files path | |||||
| ignore_urls: urls whose status response should be 404 even if the local | ignore_urls: urls whose status response should be 404 even if the local | ||||
| 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) | ||||
| if request.url in ignore_urls: | unquoted_url = unquote(request.url) | ||||
| if unquoted_url in ignore_urls: | |||||
| context.status_code = 404 | context.status_code = 404 | ||||
| return None | return None | ||||
| url = urlparse(request.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('/', '_') | ||||
| ▲ Show 20 Lines • Show All 210 Lines • Show Last 20 Lines | |||||