Merge pull request #3195 from effigies/test/fixes

TEST: Clean up tests

Author: effigies

.travis.yml

@@ -1,5 +1,5 @@
+os: linux
 dist: xenial
-sudo: true
 
 language: python
 # our build matrix
@@ -8,15 +8,13 @@ python:
 - 3.6
 - 3.7
 
-# NOTE: Any changes to the matrix section should be duplicated below for
-#       Python 3.4
 env:
   global:
     - EXTRA_WHEELS="https://5cf40426d9f06eb7461d-6fe47d9331aba7cd62fc36c7196769e4.ssl.cf2.rackcdn.com"
     - PRE_WHEELS="https://7933911d6844c6c53a7d-47bd50c35cd79bd838daf386af554a83.ssl.cf2.rackcdn.com"
     - EXTRA_PIP_FLAGS="--find-links=$EXTRA_WHEELS"
     - CHECK_TYPE=test
-  matrix:
+  jobs:
     - INSTALL_DEB_DEPENDECIES=true
       NIPYPE_EXTRAS="doc,tests,nipy,profiler"
       CI_SKIP_TEST=1
@@ -31,7 +29,7 @@ env:
       EXTRA_PIP_FLAGS="--pre $EXTRA_PIP_FLAGS --find-links $PRE_WHEELS --upgrade"
       CI_SKIP_TEST=1
 
-matrix:
+jobs:
   include:
     - python: 3.7
       env:

nipype/__init__.py

@@ -83,41 +83,9 @@ def check_latest_version(raise_exception=False):
     import etelemetry
 
     logger = logging.getLogger("nipype.utils")
-
-    INIT_MSG = "Running {packname} version {version} (latest: {latest})".format
-
-    latest = {"version": "Unknown", "bad_versions": []}
-    result = None
-    try:
-        result = etelemetry.get_project("nipy/nipype")
-    except Exception as e:
-        logger.warning("Could not check for version updates: \n%s", e)
-    finally:
-        if result:
-            latest.update(**result)
-            if LooseVersion(__version__) != LooseVersion(latest["version"]):
-                logger.info(
-                    INIT_MSG(
-                        packname="nipype", version=__version__, latest=latest["version"]
-                    )
-                )
-            else:
-                logger.info("No new version available.")
-            if latest["bad_versions"] and any(
-                [
-                    LooseVersion(__version__) == LooseVersion(ver)
-                    for ver in latest["bad_versions"]
-                ]
-            ):
-                message = (
-                    "You are using a version of Nipype with a critical "
-                    "bug. Please use a different version."
-                )
-                if raise_exception:
-                    raise RuntimeError(message)
-                else:
-                    logger.critical(message)
-    return latest
+    return etelemetry.check_available_version(
+        "nipy/nipype", __version__, logger, raise_exception
+    )
 
 
 # Run telemetry on import for interactive sessions, such as IPython, Jupyter notebooks, Python REPL

nipype/algorithms/tests/test_TSNR.py

@@ -7,7 +7,7 @@
 
 import pytest
 import numpy.testing as npt
-import mock
+from unittest import mock
 import nibabel as nb
 import numpy as np
 import os
@@ -16,9 +16,7 @@
 class TestTSNR:
     """ Note: Tests currently do a poor job of testing functionality """
 
-    in_filenames = {
-        "in_file": "tsnrinfile.nii",
-    }
+    in_filenames = {"in_file": "tsnrinfile.nii"}
 
     out_filenames = {  # default output file names
         "detrended_file": "detrend.nii.gz",

nipype/info.py

@@ -148,7 +148,7 @@ def get_nipype_gitversion():
     "simplejson>=%s" % SIMPLEJSON_MIN_VERSION,
     "traits>=%s,!=5.0" % TRAITS_MIN_VERSION,
     "filelock>=3.0.0",
-    "etelemetry",
+    "etelemetry>=0.2.0",
 ]
 
 # neurdflib has to come after prov
@@ -158,7 +158,6 @@ def get_nipype_gitversion():
 TESTS_REQUIRES = [
     "codecov",
     "coverage<5",
-    "mock",
     "pytest",
     "pytest-cov",
     "pytest-env",

nipype/interfaces/fsl/tests/test_base.py

@@ -13,8 +13,7 @@
 @pytest.mark.skipif(no_fsl(), reason="fsl is not installed")
 def test_fslversion():
     ver = fsl.Info.version()
-    ver = ver.split(".")
-    assert ver[0] in ["4", "5"]
+    assert ver.split(".", 1)[0].isdigit()
 
 
 @pytest.mark.skipif(no_fsl(), reason="fsl is not installed")

nipype/pipeline/plugins/tests/test_oar.py

@@ -1,7 +1,5 @@
 # -*- coding: utf-8 -*-
-import os
-from shutil import rmtree
-from tempfile import mkdtemp
+from shutil import which
 
 import nipype.interfaces.base as nib
 import pytest
@@ -31,13 +29,9 @@ def _list_outputs(self):
         return outputs
 
 
-@pytest.mark.xfail(reason="not known")
-def test_run_oar():
-    cur_dir = os.getcwd()
-    temp_dir = mkdtemp(prefix="test_engine_", dir=os.getcwd())
-    os.chdir(temp_dir)
-
-    pipe = pe.Workflow(name="pipe")
+@pytest.mark.skipif(which('oarsub') is None, reason="OAR not installed")
+def test_run_pbsgraph(tmp_path):
+    pipe = pe.Workflow(name="pipe", base_dir=str(tmp_path))
     mod1 = pe.Node(interface=OarTestInterface(), name="mod1")
     mod2 = pe.MapNode(interface=OarTestInterface(), iterfield=["input1"], name="mod2")
     pipe.connect([(mod1, mod2, [("output1", "input1")])])
@@ -48,5 +42,3 @@ def test_run_oar():
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]
-    os.chdir(cur_dir)
-    rmtree(temp_dir)

nipype/pipeline/plugins/tests/test_pbs.py

@@ -1,8 +1,5 @@
 # -*- coding: utf-8 -*-
-import os
-from shutil import rmtree
-from tempfile import mkdtemp
-from time import sleep
+from shutil import which
 
 import nipype.interfaces.base as nib
 import pytest
@@ -32,22 +29,15 @@ def _list_outputs(self):
         return outputs
 
 
-@pytest.mark.xfail(reason="not known")
-def test_run_pbsgraph():
-    cur_dir = os.getcwd()
-    temp_dir = mkdtemp(prefix="test_engine_")
-    os.chdir(temp_dir)
-
-    pipe = pe.Workflow(name="pipe")
+@pytest.mark.skipif(which('qsub') is None, reason="PBS not installed")
+def test_run_pbsgraph(tmp_path):
+    pipe = pe.Workflow(name="pipe", base_dir=str(tmp_path))
     mod1 = pe.Node(interface=PbsTestInterface(), name="mod1")
     mod2 = pe.MapNode(interface=PbsTestInterface(), iterfield=["input1"], name="mod2")
     pipe.connect([(mod1, mod2, [("output1", "input1")])])
-    pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="PBSGraph")
     names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]
-    os.chdir(cur_dir)
-    rmtree(temp_dir)

nipype/pipeline/plugins/tests/test_tools.py

@@ -7,7 +7,7 @@
 import scipy.sparse as ssp
 import re
 
-import mock
+from unittest import mock
 
 from nipype.pipeline.plugins.tools import report_crash
 

nipype/testing/tests/test_utils.py

@@ -7,18 +7,18 @@
 import os
 import warnings
 import subprocess
-from mock import patch, MagicMock
+from unittest.mock import patch, MagicMock
+from unittest import SkipTest
 from nipype.testing.utils import TempFATFS
 
 
 def test_tempfatfs():
     try:
         fatfs = TempFATFS()
     except (IOError, OSError):
-        warnings.warn("Cannot mount FAT filesystems with FUSE")
-    else:
-        with fatfs as tmp_dir:
-            assert os.path.exists(tmp_dir)
+        raise SkipTest("Cannot mount FAT filesystems with FUSE")
+    with fatfs as tmp_dir:
+        assert os.path.exists(tmp_dir)
 
 
 @patch(

nipype/tests/test_nipype.py

@@ -1,3 +1,4 @@
+import os
 from .. import get_info
 from ..info import get_nipype_gitversion
 import pytest
@@ -46,26 +47,28 @@ def test_no_et(tmp_path):
     from nipype.interfaces import utility as niu
     from nipype.interfaces.base import BaseInterface
 
+    et = os.getenv("NIPYPE_NO_ET") is None
+
     # Pytest doesn't trigger this, so let's pretend it's there
     with patch.object(BaseInterface, "_etelemetry_version_data", {}):
 
         # Direct function call - environment not set
         f = niu.Function(function=_check_no_et)
         res = f.run()
-        assert res.outputs.out is True
+        assert res.outputs.out == et
 
         # Basic node - environment not set
         n = pe.Node(
             niu.Function(function=_check_no_et), name="n", base_dir=str(tmp_path)
         )
         res = n.run()
-        assert res.outputs.out is True
+        assert res.outputs.out == et
 
         # Linear run - environment not set
         wf1 = pe.Workflow(name="wf1", base_dir=str(tmp_path))
         wf1.add_nodes([pe.Node(niu.Function(function=_check_no_et), name="n")])
         res = wf1.run()
-        assert next(iter(res.nodes)).result.outputs.out is True
+        assert next(iter(res.nodes)).result.outputs.out == et
 
         # MultiProc run - environment initialized with NIPYPE_NO_ET
         wf2 = pe.Workflow(name="wf2", base_dir=str(tmp_path))
@@ -91,9 +94,9 @@ def test_no_et(tmp_path):
             ]
         )
         res = wf4.run(plugin="MultiProc", plugin_args={"n_procs": 1})
-        assert next(iter(res.nodes)).result.outputs.out is True
+        assert next(iter(res.nodes)).result.outputs.out == et
 
-        # LegacyMultiProc run - environment initialized with NIPYPE_NO_ET
+        # run_without_submitting - environment not set
         wf5 = pe.Workflow(name="wf5", base_dir=str(tmp_path))
         wf5.add_nodes(
             [
@@ -105,4 +108,4 @@ def test_no_et(tmp_path):
             ]
         )
         res = wf5.run(plugin="LegacyMultiProc", plugin_args={"n_procs": 1})
-        assert next(iter(res.nodes)).result.outputs.out is True
+        assert next(iter(res.nodes)).result.outputs.out == et

nipype/utils/tests/test_config.py

@@ -5,7 +5,7 @@
 import sys
 import pytest
 from nipype import config
-from mock import MagicMock
+from unittest.mock import MagicMock
 
 try:
     import xvfbwrapper

nipype/utils/tests/test_filemanip.py

@@ -3,10 +3,9 @@
 # vi: set ft=python sts=4 ts=4 sw=4 et:
 import os
 import time
-import warnings
 from pathlib import Path
 
-import mock
+from unittest import mock, SkipTest
 import pytest
 from ...testing import TempFATFS
 from ...utils.filemanip import (
@@ -238,22 +237,22 @@ def test_copyfallback(_temp_analyze_files):
     try:
         fatfs = TempFATFS()
     except (IOError, OSError):
-        warnings.warn("Fuse mount failed. copyfile fallback tests skipped.")
-    else:
-        with fatfs as fatdir:
-            tgt_img = os.path.join(fatdir, imgname)
-            tgt_hdr = os.path.join(fatdir, hdrname)
-            for copy in (True, False):
-                for use_hardlink in (True, False):
-                    copyfile(orig_img, tgt_img, copy=copy, use_hardlink=use_hardlink)
-                    assert os.path.exists(tgt_img)
-                    assert os.path.exists(tgt_hdr)
-                    assert not os.path.islink(tgt_img)
-                    assert not os.path.islink(tgt_hdr)
-                    assert not os.path.samefile(orig_img, tgt_img)
-                    assert not os.path.samefile(orig_hdr, tgt_hdr)
-                    os.unlink(tgt_img)
-                    os.unlink(tgt_hdr)
+        raise SkipTest("Fuse mount failed. copyfile fallback tests skipped.")
+
+    with fatfs as fatdir:
+        tgt_img = os.path.join(fatdir, imgname)
+        tgt_hdr = os.path.join(fatdir, hdrname)
+        for copy in (True, False):
+            for use_hardlink in (True, False):
+                copyfile(orig_img, tgt_img, copy=copy, use_hardlink=use_hardlink)
+                assert os.path.exists(tgt_img)
+                assert os.path.exists(tgt_hdr)
+                assert not os.path.islink(tgt_img)
+                assert not os.path.islink(tgt_hdr)
+                assert not os.path.samefile(orig_img, tgt_img)
+                assert not os.path.samefile(orig_hdr, tgt_hdr)
+                os.unlink(tgt_img)
+                os.unlink(tgt_hdr)
 
 
 def test_get_related_files(_temp_analyze_files):
@@ -295,7 +294,7 @@ def test_ensure_list(filename, expected):
 
 
 @pytest.mark.parametrize(
-    "list, expected", [(["foo.nii"], "foo.nii"), (["foo", "bar"], ["foo", "bar"]),]
+    "list, expected", [(["foo.nii"], "foo.nii"), (["foo", "bar"], ["foo", "bar"])]
 )
 def test_simplify_list(list, expected):
     x = simplify_list(list)

nipype/utils/tests/test_provenance.py

@@ -2,10 +2,19 @@
 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
 # vi: set ft=python sts=4 ts=4 sw=4 et:
 import os
+from nibabel.optpkg import optional_package
+import pytest
+
+_, have_rdflib5, _ = optional_package("rdflib", min_version="5.0.0")
 
 from nipype.utils.provenance import ProvStore, safe_encode
 
+needs_rdflib5 = pytest.mark.skipif(
+    not have_rdflib5, reason="Test requires rdflib 5.0.0 or higher"
+)
+
 
+@needs_rdflib5
 def test_provenance(tmpdir):
     from nipype.interfaces.base import CommandLine
 
@@ -17,6 +26,7 @@ def test_provenance(tmpdir):
     assert "echo hello" in provn
 
 
+@needs_rdflib5
 def test_provenance_exists(tmpdir):
     tmpdir.chdir()
     from nipype import config