1from pathlib import Path
2import contextlib
3import datetime as dt
4import filelock
5import importlib
6import inspect
7import os
8import pytest
9import shutil
10import sys
11import tempfile
12import time
13import xvfbwrapper
14
15# make sure afni_test_utils importable (it will always either be not installed
16# or installed in development mode)
17sys.path.append(str(Path(__file__).parent))
18
19try:
20    import datalad.api as datalad  # noqa: F401
21
22except ImportError:
23    raise NotImplementedError("Currently datalad is a dependency for testing.")
24
25try:
26    importlib.import_module("afnipy")
27except ImportError as err:
28    # installation may be a typical "abin" install. In this case make afnipy
29    # importable fo pytest
30    bin_path = shutil.which("3dinfo")
31    if bin_path:
32        abin = Path(bin_path).parent
33        sys.path.insert(0, str(abin))
34    else:
35        raise err
36
37pytest.register_assert_rewrite("afni_test_utils.tools")
38from afni_test_utils import data_management as dm  # noqa: E40
39
40if "environ" not in inspect.signature(xvfbwrapper.Xvfb).parameters.keys():
41    raise EnvironmentError(
42        "Version of xvfbwrapper does not have the environ keyword. "
43        "Consider installing one that does. e.g. 'pip install git+git://"
44        "github.com/leej3/xvfbwrapper.git@add_support_for_xquartz_and_m"
45        "ulti_threading' "
46    )
47
48PORT_LOCKS_DIR = Path(tempfile.mkdtemp("port_locks"))
49
50cached_outdir_lock = filelock.FileLock(
51    Path("/tmp") / "session_output_directory.lock", timeout=1
52)
53
54OUTPUT_DIR_NAME_CACHE = Path("/tmp") / "test_session_outdir_name.txt"
55try:
56    cached_outdir_lock.acquire()
57    if OUTPUT_DIR_NAME_CACHE.exists():
58        OUTPUT_DIR_NAME_CACHE.unlink()
59    CURRENT_TIME = dt.datetime.strftime(dt.datetime.today(), "%Y_%m_%d_%H%M%S")
60    OUTPUT_DIR_NAME_CACHE.write_text(f"output_{CURRENT_TIME}")
61    time.sleep(5)
62    cached_outdir_lock.release()
63except filelock.Timeout:
64    OUTPUT_DIR_NAME_CACHE.read_text()
65
66encodings_patterns = [x.upper() for x in ["utf-8", "utf8"]]
67if not any(p in sys.getdefaultencoding().upper() for p in encodings_patterns):
68    raise EnvironmentError(
69        "Only utf-8 should be used for character encoding. Please "
70        "change your locale settings as required... LANG_C,LANG_ALL "
71        "etc. "
72    )
73
74
75def pytest_sessionstart(session):
76    """called after the ``Session`` object has been created and before performing collection
77    and entering the run test loop.
78
79    :param _pytest.main.Session session: the pytest session object
80    """
81    dm.get_tests_data_dir(session)
82    print("\n\n The output is being written to:", get_output_dir(session))
83
84
85def pytest_generate_tests(metafunc):
86
87    # Do some environment tweaks to homogenize behavior across systems
88    os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
89    os.environ["MPLBACKEND"] = "Agg"
90    if sys.platform == "darwin":
91        os.environ["DYLD_LIBRARY_PATH"] = "/opt/X11/lib/flat_namespace"
92
93    os.environ["AFNI_SPLASH_ANIMATE"] = "NO"
94    os.environ["NO_CMD_MOD"] = "true"
95    unset_vars = [
96        "AFNI_PLUGINPATH",
97        "AFNI_PLUGIN_PATH",
98        "BRIKCOMPRESSOR",
99        "AFNI_MODELPATH",
100        "AFNI_COMPRESSOR",
101        "AFNI_NOREALPATH",
102        "AFNI_AUTOGZIP",
103        "AFNI_GLOBAL_SESSION",
104        "AFNI_ATLAS_LIST",
105        "AFNI_TEMPLATE_SPACE_LIST",
106        "AFNI_ATLAS_PATH",
107        "AFNI_SUPP_ATLAS",
108        "AFNI_LOCAL_ATLAS",
109        "AFNI_SUPP_ATLAS_DIR",
110    ]
111    for var in unset_vars:
112        if var in os.environ:
113            del os.environ[var]
114
115    if "python_interpreter" in metafunc.fixturenames:
116        if metafunc.config.option.testpython2:
117            metafunc.parametrize("python_interpreter", ["python3", "python2"])
118        else:
119            metafunc.parametrize("python_interpreter", ["python3"])
120
121
122@pytest.fixture(scope="function")
123def data(pytestconfig, request):
124    """A function-scoped test fixture used for AFNI's testing. The fixture
125    sets up output directories as required and provides the named tuple "data"
126    to the calling function. The data object contains some fields convenient
127    for writing tests like the output directory. Finally the data fixture
128    handles test input data.files  listed in a data_paths dictionary (if
129    defined within the test module) the fixture will download them to a local
130    datalad repository as required. Paths should be listed relative to the
131    repository base-directory.
132
133    Args: request (pytest.fixture): A function level pytest request object
134        providing information about the calling test function.
135
136    Returns:
137        collections.NameTuple: A data object for conveniently handling the specification
138    """
139    data = dm.get_data_fixture(pytestconfig, request, get_output_dir(pytestconfig))
140    return data
141
142
143# configure keywords that alter test collection
144def pytest_addoption(parser):
145    parser.addoption(
146        "--runslow",
147        action="store_true",
148        default=False,
149        help="run slow tests whose execution time is on the order of many seconds)",
150    )
151    parser.addoption(
152        "--runveryslow",
153        action="store_true",
154        default=False,
155        help="run very slow tests whose execution time is on the order "
156        "of many minutes to hours ",
157    )
158    parser.addoption(
159        "--runall",
160        action="store_true",
161        default=False,
162        help="Ignore all test markers and run everything.",
163    )
164    parser.addoption(
165        "--diff-with-sample",
166        default=None,
167        help="Specify a previous tests output directory with which the output "
168        "of this test session is compared.",
169    )
170    parser.addoption(
171        "--create-sample-output",
172        action="store_true",
173        default=False,
174        help=(
175            "During many of the tests, sample output is required to "
176            "assess changes in the output files. This flag creates all "
177            "of the required files for a future comparison and no "
178            "comparison is made during the test session. "
179        ),
180    )
181    parser.addoption(
182        "--save-sample-output",
183        action="store_true",
184        default=False,
185        help=(
186            "By default, the afni_ci_test_data repository is used for "
187            "all output data comparisons during testing. This flag "
188            "updates the 'sample output' for each test run. Note that "
189            "the output that is saved may be different from the output "
190            "typically created because only files tested for "
191            "differences are included though, by default, this is all "
192            "files generated. If previous output exists, only the files "
193            "that have differences, as defined by the tests, will be "
194            "updated. Uploading updates to the publicly available "
195            "repository must be done separately. "
196        ),
197    )
198
199    parser.addoption(
200        "--testpython2",
201        action="store_true",
202        help=(
203            "For tests that use the python_interpreter fixture they are "
204            "tested in both python 3 and python 2 "
205        ),
206    )
207
208
209def pytest_collection_modifyitems(config, items):
210    """
211    This function is a pytest hook that is executed after a collection of
212    tests (but before tests are filtered using markers or keyword expression).
213
214    The current desire behavior is to filter out tests marked with slow,
215    veryslow, or combinations by default. These tests can be executed by
216    either explicitly filtering for them using a pytest marker expression or
217    by using one of the flags --runslow, --runveryslow, --runall
218    """
219    keywordexpr = config.option.keyword
220    markexpr = config.option.markexpr
221    if keywordexpr or markexpr:
222        # let pytest handle this, runveryslow or runslow flags would not have been used
223        return
224
225    if config.getoption("--runall"):
226        # marks filtered by default are included as requested by user
227        return
228
229    # define skip markers
230    skipping_veryslow_tests = not config.getoption("--runveryslow")
231    skip_veryslow = pytest.mark.skip(reason="need --runveryslow option to run")
232    skipping_slow_tests = not (
233        config.getoption("--runveryslow") or config.getoption("--runslow")
234    )
235    skip_slow = pytest.mark.skip(reason="need --runslow option to run")
236    skip_combinations = pytest.mark.skip(
237        reason="need --runall or -m 'combinations' to run"
238    )
239
240    # filter out some tests by default
241    for item in items:
242        if skipping_veryslow_tests or skipping_slow_tests:
243            if "slow" in item.keywords:
244                item.add_marker(skip_slow)
245
246        if skipping_veryslow_tests:
247            if "veryslow" in item.keywords:
248                item.add_marker(skip_veryslow)
249
250        if "combinations" in item.keywords:
251            item.add_marker(skip_combinations)
252
253
254def pytest_sessionfinish(session, exitstatus):
255    output_dir = get_output_dir(session)
256    # When configured to save output and test session was successful...
257    saving_desired = session.config.getoption("--save-sample-output")
258    user_wants = session.config.getoption("--create-sample-output")
259    create_samp_out = user_wants and not bool(exitstatus)
260    if saving_desired and not bool(exitstatus):
261        dm.save_output_to_repo()
262
263    if output_dir.exists():
264        print("\n\nTest output is written to: ", output_dir)
265
266    full_log = output_dir / "all_tests.log"
267    if full_log.exists():
268        print("\nLog is written to: ", full_log)
269
270    if create_samp_out:
271        print(
272            "\n Sample output is written to:",
273            dm.convert_to_sample_dir_path(output_dir),
274        )
275    if saving_desired and bool(exitstatus):
276        print(
277            "Sample output not saved because the test failed. You may "
278            "want to clean this up with 'cd afni_ci_test_data;git reset "
279            "--hard HEAD; git clean -df' \n Use this with caution though!"
280        )
281
282
283@pytest.fixture()
284def ptaylor_env(monkeypatch):
285    with monkeypatch.context() as m:
286        isolated_env = os.environ.copy()
287        isolated_env["AFNI_COMPRESSOR"] = "GZIP"
288        m.setattr(os, "environ", isolated_env)
289        try:
290            yield
291        finally:
292            pass
293
294
295@pytest.fixture()
296def unique_gui_port():
297
298    PORT_NUM = 0
299    while True:
300        try:
301            PORT_NUM += 1
302            gui_port_lock_path = PORT_LOCKS_DIR / f"gui_port_{PORT_NUM}.lock"
303            GUI_PORT_LOCK = filelock.FileLock(gui_port_lock_path)
304            GUI_PORT_LOCK.acquire(timeout=1, poll_intervall=0.5)
305            return PORT_NUM
306        except filelock.Timeout:
307            continue
308
309
310@pytest.fixture(autouse=True)
311def _use_test_dir(request):
312    tests_dir = Path(__file__).parent
313    with working_directory(tests_dir):
314        yield
315
316
317@contextlib.contextmanager
318def working_directory(path):
319    """Changes working directory and returns to previous on exit."""
320    prev_cwd = Path.cwd()
321    os.chdir(path)
322    try:
323        yield
324    finally:
325        os.chdir(prev_cwd)
326
327
328@pytest.fixture()
329def mocked_abin():
330    """
331    Fixture to supply 'mocked_abin', a directory containing trivial
332    executables called 3dinfo and align_epi_anat.py and afnipy/afni_base.py (an
333    importable python module)
334    """
335    temp_dir = tempfile.mkdtemp()
336    abin_dir = Path(temp_dir) / "abin"
337    abin_dir.mkdir()
338
339    (abin_dir / "3dinfo").touch(mode=0o777)
340    (abin_dir / "libmri.so").touch(mode=0o444)
341    (abin_dir / "3dinfo").write_text("#!/usr/bin/env bash\necho success")
342    (abin_dir / "align_epi_anat.py").touch(mode=0o777)
343    (abin_dir / "align_epi_anat.py").write_text("#!/usr/bin/env bash\necho success")
344    (abin_dir / "afnipy").mkdir()
345    (abin_dir / "afnipy/afni_base.py").touch()
346    (abin_dir / "afnipy/__init__.py").touch()
347    return abin_dir
348
349
350@pytest.fixture()
351def mocked_hierarchical_installation():
352    """
353    Creates a fake installation directory containing trivial executables
354    called 3dinfo and align_epi_anat.py and afnipy/afni_base.py (an importable
355    python module) along with libmri in the appropriate organization defined
356    by the GNU installation guidelines. To use this you should add the
357    fake_site_packages directory to sys.path to simulate the afnipy package
358    being installed into the current interpreter.
359    """
360    temp_dir = tempfile.mkdtemp()
361    # Create an installation directory
362    instd = Path(temp_dir) / "abin_hierarchical"
363    instd.mkdir()
364    bindir = instd / "bin"
365    bindir.mkdir()
366
367    (bindir / "3dinfo").touch(mode=0o777)
368    (bindir / "3dinfo").write_text("#!/usr/bin/env bash\necho success")
369    (instd / "lib/").mkdir()
370    (instd / "lib/libmri.so").touch(mode=0o444)
371    (instd / "fake_site_packages").mkdir()
372    (instd / "fake_site_packages/align_epi_anat.py").touch(mode=0o777)
373    (instd / "fake_site_packages/align_epi_anat.py").write_text(
374        "#!/usr/bin/env bash\necho success"
375    )
376    (instd / "fake_site_packages/afnipy").mkdir()
377    (instd / "fake_site_packages/afnipy/afni_base.py").touch()
378    (instd / "fake_site_packages/afnipy/__init__.py").touch()
379    return instd
380
381
382def get_output_dir(config_obj):
383    if hasattr(config_obj, "config"):
384        conf = config_obj.config
385    elif hasattr(config_obj, "rootdir"):
386        conf = config_obj
387    else:
388        print(config_obj)
389        raise TypeError("A pytest config object was expected")
390
391    rootdir = conf.rootdir
392
393    output_dir = Path(rootdir) / "output_of_tests" / OUTPUT_DIR_NAME_CACHE.read_text()
394    if not output_dir.exists():
395        output_dir.mkdir(parents=True)
396
397    return output_dir
398