diff --git a/swh/loader/package/tests/conftest.py b/swh/loader/package/tests/conftest.py index 4ba6bff..0e4c7bc 100644 --- a/swh/loader/package/tests/conftest.py +++ b/swh/loader/package/tests/conftest.py @@ -1,122 +1,124 @@ # Copyright (C) 2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import logging import os import re import pytest from functools import partial from os import path from urllib.parse import urlparse from .common import DATADIR logger = logging.getLogger(__name__) # Check get_local_factory function # Maximum number of iteration checks to generate requests responses MAX_VISIT_FILES = 10 @pytest.fixture def swh_config(monkeypatch): conffile = os.path.join(DATADIR, 'loader.yml') monkeypatch.setenv('SWH_CONFIG_FILENAME', conffile) return conffile -def get_response_cb(request, context, ignore_urls=[], visit=None): +def get_response_cb(request, context, ignore_urls=[], visits=None): """Mount point callback to fetch on disk the content of a request This is meant to be used as 'body' argument of the requests_mock.get() method. It will look for files on the local filesystem based on the requested URL, using the following rules: - files are searched in the DATADIR/ directory - the local file name is the path part of the URL with path hierarchy markers (aka '/') replaced by '_' Eg. if you use the requests_mock fixture in your test file as: requests_mock.get('https://nowhere.com', body=get_response_cb) # or even requests_mock.get(re.compile('https://'), body=get_response_cb) then a call requests.get like: requests.get('https://nowhere.com/path/to/resource') will look the content of the response in: DATADIR/resources/nowhere.com/path_to_resource Args: request (requests.Request): Object requests context (requests.Context): Object holding response metadata - information (status_code, headers, etc...) + information (status_code, headers, etc...) ignore_urls (List): urls whose status response should be 404 even if - the local file exists - visit (Optional[int]): Visit number for the given url (can be None) + the local file exists + visits (Optional[Dict]): Map of url, number of visits. If None, disable + multi visit support (default) Returns: Optional[FileDescriptor] on the on disk file to read from the test context """ logger.debug('get_response_cb(%s, %s)', request, context) logger.debug('url: %s', request.url) logger.debug('ignore_urls: %s', ignore_urls) if request.url in ignore_urls: context.status_code = 404 return None url = urlparse(request.url) dirname = url.hostname # pypi.org | files.pythonhosted.org # url.path: pypi//json -> local file: pypi__json filename = url.path[1:] if filename.endswith('/'): filename = filename[:-1] filename = filename.replace('/', '_') filepath = path.join(DATADIR, dirname, filename) - if visit: - filepath = filepath + '_visit%s' % visit + if visits is not None: + visit = visits.get(url, 0) + visits[url] = visit + 1 + if visit: + filepath = filepath + '_visit%s' % visit + if not path.isfile(filepath): + logger.debug('not found filepath: %s', filepath) context.status_code = 404 return None fd = open(filepath, 'rb') context.headers['content-length'] = str(path.getsize(filepath)) return fd def local_get_factory(ignore_urls=[], has_multi_visit=False): @pytest.fixture def local_get(requests_mock): if not has_multi_visit: cb = partial(get_response_cb, ignore_urls=ignore_urls) requests_mock.get(re.compile('https://'), body=cb) else: - requests_mock.get(re.compile('https'), [ - { - 'body': partial( - get_response_cb, - ignore_urls=ignore_urls, - visit=i) - } for i in range(MAX_VISIT_FILES)] + visits = {} + requests_mock.get(re.compile('https'), body=partial( + get_response_cb, ignore_urls=ignore_urls, visits=visits) ) return requests_mock return local_get local_get = local_get_factory([]) local_get_visits = local_get_factory(has_multi_visit=True) diff --git a/swh/loader/package/tests/resources/example.com/other.json b/swh/loader/package/tests/resources/example.com/other.json new file mode 100644 index 0000000..14e24d4 --- /dev/null +++ b/swh/loader/package/tests/resources/example.com/other.json @@ -0,0 +1 @@ +"foobar" diff --git a/swh/loader/package/tests/test_conftest.py b/swh/loader/package/tests/test_conftest.py index d2173bc..dce8a41 100644 --- a/swh/loader/package/tests/test_conftest.py +++ b/swh/loader/package/tests/test_conftest.py @@ -1,61 +1,83 @@ # Copyright (C) 2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import requests from swh.loader.package.tests.conftest import local_get_factory +def test_get_response_cb_with_visits_nominal(local_get_visits): + response = requests.get('https://example.com/file.json') + assert response.ok + assert response.json() == {'hello': 'you'} + + response = requests.get('https://example.com/file.json') + assert response.ok + assert response.json() == {'hello': 'world'} + + response = requests.get('https://example.com/file.json') + assert not response.ok + assert response.status_code == 404 + + def test_get_response_cb_with_visits(local_get_visits): response = requests.get('https://example.com/file.json') assert response.ok assert response.json() == {'hello': 'you'} + response = requests.get('https://example.com/other.json') + assert response.ok + assert response.json() == "foobar" + response = requests.get('https://example.com/file.json') assert response.ok assert response.json() == {'hello': 'world'} + response = requests.get('https://example.com/other.json') + assert not response.ok + assert response.status_code == 404 + response = requests.get('https://example.com/file.json') assert not response.ok assert response.status_code == 404 def test_get_response_cb_no_visit(local_get): response = requests.get('https://example.com/file.json') assert response.ok assert response.json() == {'hello': 'you'} response = requests.get('https://example.com/file.json') assert response.ok assert response.json() == {'hello': 'you'} local_get_ignore = local_get_factory( ignore_urls=['https://example.com/file.json'], has_multi_visit=False, ) def test_get_response_cb_ignore_url(local_get_ignore): response = requests.get('https://example.com/file.json') assert not response.ok assert response.status_code == 404 local_get_ignore_and_visit = local_get_factory( ignore_urls=['https://example.com/file.json'], has_multi_visit=True, ) def test_get_response_cb_ignore_url_with_visit(local_get_ignore_and_visit): response = requests.get('https://example.com/file.json') assert not response.ok assert response.status_code == 404 response = requests.get('https://example.com/file.json') assert not response.ok assert response.status_code == 404