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 20 Lines • Show All 89 Lines • ▼ Show 20 Lines | definition in the local conftest, for example:: | ||||
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, | |||||
datadir=datadir) | |||||
requests_mock.get(re.compile('https?://'), body=cb) | 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 | ||||
▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Lines | def build_response(self, req, resp): | ||||
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 | ||||
Show All 22 Lines |