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