Add pytest fixture to allow mocking requests with data file

Roughly, the idea is that http queries are transformed into file queries within
http context. If found, the data file is served or not (404). This translates
into query response so that the code can actually complete as usual.

For technical reasons the pytest plugin is in a dedicated non-swh module. This
is actually required for the pytest plugin loading mecanism to work without
hijacking the loading of swh packages. See [1] for more details.

[1] https://github.com/pytest-dev/pytest/issues/2042

MANIFEST.in

@@ -6,3 +6,5 @@ include requirements-db.txt
 include version.txt
 recursive-include swh/core/sql *.sql
 recursive-include swh py.typed
+recursive-include swh/core/tests/data/ *
+recursive-include swh/core/tests/fixture/data/ *

setup.py

@@ -46,6 +46,7 @@ setup(
     author_email='swh-devel@inria.fr',
     url='https://forge.softwareheritage.org/diffusion/DCORE/',
     packages=find_packages(),
+    py_modules=['pytest_swh_core'],
     scripts=[],
     install_requires=parse_requirements(None, 'swh'),
     setup_requires=['vcversioner'],
@@ -63,6 +64,8 @@ setup(
         [swh.cli.subcommands]
         db=swh.core.cli.db:db
         db-init=swh.core.cli.db:db_init
+        [pytest11]
+        pytest_swh_core = swh.core.pytest_plugin
     ''',
     classifiers=[
         "Programming Language :: Python :: 3",

swh/core/pytest_plugin.py

+# 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 re
+import pytest
+
+from functools import partial
+from os import path
+from typing import Dict, List, Optional
+from urllib.parse import urlparse
+
+
+logger = logging.getLogger(__name__)
+
+
+# Check get_local_factory function
+# Maximum number of iteration checks to generate requests responses
+MAX_VISIT_FILES = 10
+
+
+def get_response_cb(request, context, datadir,
+                    ignore_urls: List[str] = [],
+                    visits: Optional[Dict] = None):
+    """Mount point callback to fetch on disk the request's content.
+
+    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/<hostname> 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/nowhere.com/path_to_resource
+
+    Args:
+        request (requests.Request): Object requests
+        context (requests.Context): Object holding response metadata
+            information (status_code, headers, etc...)
+        ignore_urls: urls whose status response should be 404 even if the local
+            file exists
+        visits: Dict of url, number of visits. If None, disable multi visit
+            support (default)
+
+    Returns:
+        Optional[FileDescriptor] 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/<project>/json -> local file: pypi_<project>_json
+    filename = url.path[1:]
+    if filename.endswith('/'):
+        filename = filename[:-1]
+    filename = filename.replace('/', '_')
+    filepath = path.join(datadir, dirname, filename)
+    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
+
+
+@pytest.fixture
+def datadir(request):
+    """By default, returns the test directory's data directory.
+
+    This can be overriden on a per arborescence basis. Add an override
+    definition in the local conftest, for example:
+
+        import pytest
+
+        from os import path
+
+        @pytest.fixture
+        def datadir():
+            return path.join(path.abspath(path.dirname(__file__)), 'resources')
+
+
+    """
+    return path.join(path.dirname(str(request.fspath)), 'data')
+
+
+def requests_mock_datadir_factory(ignore_urls: List[str] = [],
+                                  has_multi_visit: bool = False):
+    """This factory generates fixture which allow to look for files on the
+    local filesystem based on the requested URL, using the following rules:
+
+    - files are searched in the datadir/<hostname> directory
+
+    - the local file name is the path part of the URL with path hierarchy
+      markers (aka '/') replaced by '_'
+
+    Multiple implementations are possible, for example:
+
+    - requests_mock_datadir_factory([]):
+        This computes the file name from the query and always returns the same
+        result.
+
+    - requests_mock_datadir_factory(has_multi_visit=True):
+        This computes the file name from the query and returns the content of
+        the filename the first time, the next call returning the content of
+        files suffixed with _visit1 and so on and so forth. If the file is not
+        found, returns a 404.
+
+    - requests_mock_datadir_factory(ignore_urls=['url1', 'url2']):
+        This will ignore any files corresponding to url1 and url2, always
+        returning 404.
+
+    Args:
+        ignore_urls: List of urls to always returns 404 (whether file
+            exists or not)
+        has_multi_visit: Activate or not the multiple visits behavior
+
+    """
+    @pytest.fixture
+    def requests_mock_datadir(requests_mock, datadir):
+        if not has_multi_visit:
+            cb = partial(get_response_cb,
+                         ignore_urls=ignore_urls,
+                         datadir=datadir)
+            requests_mock.get(re.compile('https://'), body=cb)
+        else:
+            visits = {}
+            requests_mock.get(re.compile('https://'), body=partial(
+                get_response_cb, ignore_urls=ignore_urls, visits=visits,
+                datadir=datadir)
+            )
+
+        return requests_mock
+
+    return requests_mock_datadir
+
+
+# Default `requests_mock_datadir` implementation
+requests_mock_datadir = requests_mock_datadir_factory([])
+
+# Implementation for multiple visits behavior:
+# - first time, it checks for a file named `filename`
+# - second time, it checks for a file named `filename`_visit1
+# etc...
+requests_mock_datadir_visits = requests_mock_datadir_factory(
+    has_multi_visit=True)

swh/core/tests/data/example.com/file.json

+{
+    "hello": "you"
+}

swh/core/tests/data/example.com/file.json_visit1

+{
+    "hello": "world"
+}

swh/core/tests/data/example.com/other.json

+"foobar"

swh/core/tests/fixture/conftest.py

+# 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 pytest
+
+from os import path
+
+
+DATADIR = path.join(path.abspath(path.dirname(__file__)), 'data')
+
+
+@pytest.fixture
+def datadir():
+    return DATADIR

swh/core/tests/fixture/data/example.com/file.json

+{
+    "welcome": "you"
+}

swh/core/tests/fixture/test_pytest_plugin.py

+# 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 .conftest import DATADIR
+
+# In this arborescence, we override in the local conftest.py module the
+# "datadir" fixture to specify where to retrieve the data files from.
+
+
+def test_requests_mock_datadir_with_datadir_fixture_override(
+        requests_mock_datadir):
+    """Override datadir fixture should retrieve data from elsewhere
+
+    """
+    response = requests.get('https://example.com/file.json')
+    assert response.ok
+    assert response.json() == {'welcome': 'you'}
+
+
+def test_data_dir_override(datadir):
+    assert datadir == DATADIR

swh/core/tests/test_pytest_plugin.py

+# 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 os import path
+
+from swh.core.pytest_plugin import requests_mock_datadir_factory
+
+
+def test_get_response_cb_with_visits_nominal(requests_mock_datadir_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(requests_mock_datadir_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(requests_mock_datadir):
+    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'}
+
+
+requests_mock_datadir_ignore = requests_mock_datadir_factory(
+    ignore_urls=['https://example.com/file.json'],
+    has_multi_visit=False,
+)
+
+
+def test_get_response_cb_ignore_url(requests_mock_datadir_ignore):
+    response = requests.get('https://example.com/file.json')
+    assert not response.ok
+    assert response.status_code == 404
+
+
+requests_mock_datadir_ignore_and_visit = requests_mock_datadir_factory(
+    ignore_urls=['https://example.com/file.json'],
+    has_multi_visit=True,
+)
+
+
+def test_get_response_cb_ignore_url_with_visit(
+        requests_mock_datadir_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
+
+
+def test_data_dir(datadir):
+    expected_datadir = path.join(path.abspath(path.dirname(__file__)), 'data')
+    assert datadir == expected_datadir