"""Collection of fixtures for facilitation test implementations
"""
import logging
import os
from pathlib import Path
import subprocess
import pytest
from tempfile import NamedTemporaryFile
from time import sleep
from unittest.mock import patch
from urllib.request import urlopen
from datalad_next.datasets import Dataset
from datalad_next.tests.utils import (
HTTPPath,
SkipTest,
WebDAVPath,
external_versions,
get_git_config_global_fpath,
md5sum,
)
lgr = logging.getLogger('datalad.next.tests.fixtures')
[docs]@pytest.fixture(autouse=False, scope="function")
def tmp_keyring():
"""Patch plaintext keyring to temporarily use a different storage
No credential read or write actions will impact any existing credential
store of any configured backend.
The patched backend is yielded by the fixture.
"""
import keyring
# the testsetup assumes this to be a plaintext backend.
# this backend is based on a filename and maintains no state.
# each operation opens, reads/writes, and then closes the file.
# hence we can simply point to a different file
backend = keyring.get_keyring()
prev_fpath = backend.file_path
# no tmp keyring yet, make one
with NamedTemporaryFile(
'r',
prefix='datalad_tmp_keyring_',
delete=True) as tf:
# we must close, because windows does not like the file being open
# already when ConfigManager would open it for reading
tf.close()
backend.file_path = tf.name
with patch.dict(os.environ,
{'DATALAD_TESTS_TMP_KEYRING_PATH': tf.name}):
yield backend
backend.file_path = prev_fpath
# the following is taken from datalad/conftest.py
# sadly, this is defined inline and cannot be reused directly
standard_gitconfig = """\
[user]
name = DataLad Tester
email = test@example.com
[core]
askPass =
[datalad "log"]
exc = 1
[annex "security"]
# from annex 6.20180626 file:/// and http://localhost access isn't
# allowed by default
allowed-url-schemes = http https file
allowed-http-addresses = all
[protocol "file"]
# since git 2.38.1 cannot by default use local clones for submodules
# https://github.blog/2022-10-18-git-security-vulnerabilities-announced/#cve-2022-39253
allow = always
""" + os.environ.get('DATALAD_TESTS_GITCONFIG', '').replace('\\n', os.linesep)
[docs]@pytest.fixture(autouse=False, scope="function")
def datalad_cfg():
"""Temporarily alter configuration to use a plain "global" configuration
The global configuration manager at `datalad.cfg` is reloaded after
adjusting `GIT_CONFIG_GLOBAL` to point to a new temporary `.gitconfig`
file.
After test execution the file is removed, and the global `ConfigManager`
is reloaded once more.
Any test using this fixture will be skipped for Git versions earlier
than 2.32, because the `GIT_CONFIG_GLOBAL` environment variable used
here was only introduced with that version.
"""
if external_versions['cmd:git'] < "2.32":
raise SkipTest(
"Git configuration redirect via GIT_CONFIG_GLOBAL "
"only supported since Git v2.32"
)
from datalad import cfg
with NamedTemporaryFile(
'w',
prefix='datalad_gitcfg_global_',
delete=False) as tf:
tf.write(standard_gitconfig)
# we must close, because windows does not like the file being open
# already when ConfigManager would open it for reading
tf.close()
with patch.dict(os.environ, {'GIT_CONFIG_GLOBAL': tf.name}):
cfg.reload(force=True)
yield cfg
# reload to put the previous config in effect again
cfg.reload(force=True)
[docs]@pytest.fixture(autouse=True, scope="function")
def check_gitconfig_global():
"""No test must modify a user's global Git config.
If such modifications are needed, a custom configuration setup
limited to the scope of the test requiring it must be arranged.
"""
globalcfg_fname = get_git_config_global_fpath()
if not globalcfg_fname.exists():
lgr.warning(
'No global/user Git config file exists. This is an unexpected '
'test environment, no config modifications checks can be '
'performed. Proceeding nevertheless.')
# let the test run
yield
# and exit quietly
return
# we have a config file. hash it pre and post test. Fail is changed.
pre = md5sum(globalcfg_fname)
yield
post = md5sum(globalcfg_fname)
assert pre == post, \
"Global Git config modification detected. Test must be modified to use " \
"a temporary configuration target. Hint: use the `datalad_cfg` fixture."
[docs]@pytest.fixture(autouse=True, scope="function")
def check_plaintext_keyring():
"""No test must modify a user's keyring.
If such modifications are needed, a custom keyring setup
limited to the scope of the test requiring it must be arranged.
The ``tmp_keyring`` fixture can be employed in such cases.
"""
# datalad-core configures keyring to use a plaintext backend
# we will look for the underlying file and verify that it is either
# no there, or remains unmodified
import keyring
kr = keyring.get_keyring()
if not hasattr(kr, 'file_path'):
# this is not the plain text keyring, nothing we can do here
# run as-is, but leave a message
lgr.warning('Running without the expected plain-text keyring')
yield
return
kr_fpath = Path(kr.file_path)
pre = md5sum(kr_fpath) if kr_fpath.exists() else ''
yield
post = md5sum(kr_fpath) if kr_fpath.exists() else ''
assert pre == post, \
"Keyring modification detected. Test must be modified to use " \
"a temporary keyring. Hint: use the `tmp_keyring` fixture."
[docs]@pytest.fixture(autouse=False, scope="function")
def credman(datalad_cfg, tmp_keyring):
"""Provides a temporary credential manager
It comes with a temporary global datalad config and a temporary
keyring as well.
This manager can be used to deploy or manipulate credentials within the
scope of a single test.
"""
from datalad import cfg
from datalad_next.credman import CredentialManager
cm = CredentialManager(cfg)
yield cm
[docs]@pytest.fixture(autouse=False, scope="function")
def dataset(datalad_cfg, tmp_path_factory):
"""Provides a ``Dataset`` instance for a not-yet-existing repository
The instance points to an existing temporary path, but ``create()``
has not been called on it yet.
"""
# must use the factory to get a unique path even when a concrete
# test also uses `tmp_path`
ds = Dataset(tmp_path_factory.mktemp("dataset"))
yield ds
[docs]@pytest.fixture(autouse=False, scope="function")
def existing_dataset(dataset):
"""Provides a ``Dataset`` instance pointing to an existing dataset/repo
This fixture uses an instance provided by the ``dataset`` fixture and
calls ``create()`` on it, before it yields the ``Dataset`` instance.
"""
dataset.create(result_renderer='disabled')
yield dataset
[docs]@pytest.fixture(autouse=False, scope="function")
def existing_noannex_dataset(dataset):
"""just like ``existing_dataset``, but created with ``annex=False``
"""
dataset.create(annex=False, result_renderer='disabled')
yield dataset
[docs]@pytest.fixture(autouse=False, scope="session")
def webdav_credential():
"""Provides HTTP Basic authentication credential necessary to access the
server provided by the ``webdav_server`` fixture."""
yield dict(
name='dltest-my&=webdav',
user='datalad',
secret='secure',
type='user_password',
)
[docs]@pytest.fixture(autouse=False, scope="function")
def webdav_server(tmp_path_factory, webdav_credential):
"""Provides a WebDAV server, serving a temporary directory
The fixtures yields an instance of ``WebDAVPath``, providing the
following essential attributes:
- ``path``: ``Path`` instance of the served temporary directory
- ``url``: HTTP URL to access the WebDAV server
Server access requires HTTP Basic authentication with the credential
provided by the ``webdav_credential`` fixture.
"""
auth = (webdav_credential['user'], webdav_credential['secret'])
# must use the factory to get a unique path even when a concrete
# test also uses `tmp_path`
path = tmp_path_factory.mktemp("webdav")
# this looks a little awkward, but is done to avoid a change in
# WebDAVPath.
server = WebDAVPath(path, auth=auth)
with server as server_url:
server.url = server_url
yield server
[docs]@pytest.fixture(autouse=False, scope="session")
def http_credential():
"""Provides the HTTP Basic authentication credential necessary to access the
HTTP server provided by the ``http_server_with_basicauth`` fixture."""
yield dict(
name='dltest-my&=http',
user='datalad',
secret='secure',
type='user_password',
)
[docs]@pytest.fixture(autouse=False, scope="function")
def http_server(tmp_path_factory):
"""Provides an HTTP server, serving a temporary directory
The fixtures yields an instance of ``HTTPPath``, providing the
following essential attributes:
- ``path``: ``Path`` instance of the served temporary directory
- ``url``: HTTP URL to access the HTTP server
"""
# must use the factory to get a unique path even when a concrete
# test also uses `tmp_path`
path = tmp_path_factory.mktemp("webdav")
server = HTTPPath(path, use_ssl=False, auth=None)
with server:
# overwrite path with Path object for convenience
server.path = path
yield server
[docs]@pytest.fixture(autouse=False, scope="function")
def http_server_with_basicauth(tmp_path_factory, http_credential):
"""Like ``http_server`` but requiring authentication via ``http_credential``
"""
path = tmp_path_factory.mktemp("webdav")
server = HTTPPath(
path, use_ssl=False,
auth=(http_credential['user'], http_credential['secret']),
)
with server:
# overwrite path with Path object for convenience
server.path = path
yield server
[docs]@pytest.fixture(scope="session")
def httpbin_service():
"""Return canonical access URLs for the HTTPBIN service
This fixture tries to spin up a httpbin Docker container at localhost:8765;
if successful, it returns this URL as the 'standard' URL. If the attempt
fails, a URL pointing to the canonical instance is returned.
For tests that need to have the service served via a specific
protocol (https vs http), the corresponding URLs are returned
too. They always point to the canonical deployment, as some
tests require both protocols simultaneously and a local deployment
generally won't have https.
"""
hburl = 'http://httpbin.org'
hbsurl = 'https://httpbin.org'
ciurl = 'http://localhost:8765'
if os.name == "posix":
try:
r = subprocess.run(
["docker", "run", "-d", "-p", "127.0.0.1:8765:80", "kennethreitz/httpbin"],
check=True,
stdout=subprocess.PIPE,
text=True,
)
except (OSError, subprocess.CalledProcessError):
lgr.warning("Failed to spin up httpbin Docker container:", exc_info=True)
container_id = None
else:
container_id = r.stdout.strip()
else:
container_id = None
try:
if container_id is not None:
# Wait for container to fully start:
for _ in range(25):
try:
urlopen(ciurl)
except Exception:
sleep(1)
else:
break
else:
raise RuntimeError("httpbin container did not start up in time")
yield {
"standard": ciurl if container_id is not None else hbsurl,
"http": hburl,
"https": hbsurl,
}
finally:
if container_id is not None:
subprocess.run(["docker", "rm", "-f", container_id], check=True)
[docs]@pytest.fixture(scope="function")
def httpbin(httpbin_service):
"""Does the same thing as ``httpbin_service``, but skips on function-scope
``httpbin_service`` always returns access URLs for HTTPBIN. However,
in some cases it is simply not desirable to run a test. For example,
the appveyor workers are more or less constantly unable to access the
public service. This fixture is evaluated at function-scope and
raises ``SkipTest`` whenever any of these undesired conditions is
detected. Otherwise it just relays ``httpbin_service``.
"""
if os.environ.get('DATALAD_TESTS_NONETWORK'):
raise SkipTest(
'Not running httpbin-based test: NONETWORK flag set'
)
if 'APPVEYOR' in os.environ and 'DEPLOY_HTTPBIN_IMAGE' not in os.environ:
raise SkipTest(
"Not running httpbin-based test on appveyor without "
"docker-deployed instance -- too unreliable"
)
yield httpbin_service
[docs]@pytest.fixture(autouse=False, scope="function")
def datalad_interactive_ui(monkeypatch):
"""Yields a UI replacement to query for operations and stage responses
No output will be written to STDOUT/ERR by this UI.
A standard usage pattern is to stage one or more responses, run the
to-be-tested code, and verify that the desired user interaction
took place::
> datalad_interactive_ui.staged_responses.append('skip')
> ...
> assert ... datalad_interactive_ui.log
"""
from datalad_next.uis import ui_switcher
from datalad_next.tests.utils import InteractiveTestUI
with monkeypatch.context() as m:
m.setattr(ui_switcher, '_ui', InteractiveTestUI())
yield ui_switcher.ui
[docs]@pytest.fixture(autouse=False, scope="function")
def datalad_noninteractive_ui(monkeypatch):
"""Yields a UI replacement to query for operations
No output will be written to STDOUT/ERR by this UI.
A standard usage pattern is to run the to-be-tested code, and verify that
the desired user messaging took place::
> ...
> assert ... datalad_interactive_ui.log
"""
from datalad_next.uis import ui_switcher
from datalad_next.tests.utils import TestUI
with monkeypatch.context() as m:
m.setattr(ui_switcher, '_ui', TestUI())
yield ui_switcher.ui