debugged running of generated of unittests in nipype2pydra unittests

Author: tclose

nipype2pydra/cli.py

@@ -34,7 +34,8 @@ def cli():
     help="a Python file containing callable functions required in the command interface",
 )
 @click.option(
-    "--output-module" "-m",
+    "--output-module",
+    "-m",
     type=str,
     default=None,
     help=(

nipype2pydra/task.py

@@ -39,7 +39,11 @@ def output_callables_validator(self, _, output_callables: dict):
     def __attrs_post_init__(self):
         if self.output_module is None:
             if self.nipype_module.__name__.startswith("nipype.interfaces."):
-                self.output_module = self.nipype_module.__name__[len("nipype.interfaces."):]
+                self.output_module = (
+                    "pydra.tasks."
+                    + self.nipype_module.__name__[len("nipype.interfaces.") :]
+                    + "." + self.task_name.lower()
+                )
             else:
                 raise RuntimeError(
                     "Output-module needs to be explicitly provided to task converter "
@@ -67,14 +71,14 @@ def generate(self, package_root: Path):
         input_fields, inp_templates = self.convert_input_fields()
         output_fields = self.convert_output_spec(fields_from_template=inp_templates)
 
-        module_path = (Path(package_root) / "pydra" / "tasks").joinpath(self.output_module.split("."))
-        output_file = (module_path / self.task_name.lower()).with_suffix(".py")
+        output_file = Path(package_root).joinpath(*self.output_module.split(".")).with_suffix(".py")
+        testdir = output_file.parent / "tests"
+        testdir.mkdir(parents=True)
+
         self.write_task(output_file, input_fields, output_fields)
 
-        testdir = module_path / "tests"
-        testdir.mkdir()
-        filename_test = testdir / f"test_spec_{self.task_name.lower()}"
-        filename_test_run = testdir / f"test_run_{self.task_name.lower()}"
+        filename_test = testdir / f"test_spec_{self.task_name.lower()}.py"
+        filename_test_run = testdir / f"test_run_{self.task_name.lower()}.py"
         self.write_test(filename_test=filename_test)
         self.write_test(filename_test=filename_test_run, run=True)
 
@@ -359,9 +363,7 @@ def write_test(self, filename_test, run=False):
                 tests_inp_error.append((tests_inputs[i], out))
 
         spec_str = "import os, pytest \nfrom pathlib import Path\n"
-        spec_str += (
-            f"from pydra.tasks.{self.output_module}.{self.task_name.lower()} import {self.task_name} \n\n"
-        )
+        spec_str += f"from {self.output_module} import {self.task_name} \n\n"
         if run:
             pass
         spec_str += f"@pytest.mark.parametrize('inputs, outputs', {tests_inp_outp})\n"
@@ -399,9 +401,7 @@ def write_test_error(self, input_error):
         """
         spec_str = "\n\n"
         spec_str += f"@pytest.mark.parametrize('inputs, error', {input_error})\n"
-        spec_str += (
-            f"def test_{self.task_name}_exception(test_data, inputs, error):\n"
-        )
+        spec_str += f"def test_{self.task_name}_exception(test_data, inputs, error):\n"
         spec_str += "    in_file = Path(test_data) / 'test.nii.gz'\n"
         spec_str += "    if inputs is None: inputs = {{}}\n"
         spec_str += "    for key, val in inputs.items():\n"

nipype2pydra/utils.py

@@ -1,7 +1,10 @@
 import traceback
 from types import ModuleType
 import sys
+import os
+from contextlib import contextmanager
 from pathlib import Path
+
 from importlib import import_module
 
 
@@ -24,3 +27,38 @@ def import_module_from_path(module_path: Path) -> ModuleType:
         return import_module(module_path.stem)
     finally:
         sys.path.pop(0)
+
+
+@contextmanager
+def set_cwd(path):
+    """Sets the current working directory to `path` and back to original
+    working directory on exit
+
+    Parameters
+    ----------
+    path : str
+        The file system path to set as the current working directory
+    """
+    pwd = os.getcwd()
+    os.chdir(path)
+    try:
+        yield path
+    finally:
+        os.chdir(pwd)
+
+
+@contextmanager
+def add_to_sys_path(path: Path):
+    """Adds the given `path` to the Python system path and then reverts it back to the
+    original value on exit
+
+    Parameters
+    ----------
+    path : str
+        The file system path to add to the system path
+    """
+    sys.path.insert(0, str(path))
+    try:
+        yield sys.path
+    finally:
+        sys.path.pop(0)

tests/test_task.py

@@ -1,8 +1,13 @@
 from importlib import import_module
 import yaml
 from conftest import show_cli_trace
+import pytest
+import logging
 from nipype2pydra.cli import task as task_cli
-from nipype2pydra.utils import import_module_from_path
+from nipype2pydra.utils import add_to_sys_path
+
+
+logging.basicConfig(level=logging.INFO)
 
 
 INBUILT_NIPYPE_TRAIT_NAMES = [
@@ -18,19 +23,24 @@ def test_task_conversion(task_spec_file, cli_runner, work_dir):
 
     with open(task_spec_file) as f:
         task_spec = yaml.safe_load(f)
-    output_file = (work_dir / task_spec_file.stem).with_suffix(".py")
+    pkg_root = work_dir / "src"
+
+    output_module_path = f"nipype2pydratest.{task_spec_file.stem.lower()}"
 
     result = cli_runner(
         task_cli,
         args=[
             str(task_spec_file),
-            str(output_file),
+            str(pkg_root),
+            "--output-module",
+            output_module_path,
         ],
     )
 
     assert result.exit_code == 0, show_cli_trace(result)
 
-    pydra_module = import_module_from_path(output_file)
+    with add_to_sys_path(pkg_root):
+        pydra_module = import_module(output_module_path)
     pydra_task = getattr(pydra_module, task_spec["task_name"])
     nipype_interface = getattr(
         import_module(task_spec["nipype_module"]), task_spec["task_name"]
@@ -43,8 +53,15 @@ def test_task_conversion(task_spec_file, cli_runner, work_dir):
         for n in nipype_trait_names
         if not (
             n in INBUILT_NIPYPE_TRAIT_NAMES
-            or (n.endswith("_items") and n[:-len("_items")] in nipype_trait_names)
+            or (n.endswith("_items") and n[: -len("_items")] in nipype_trait_names)
         )
     )
 
-    # TODO: More detailed tests needed here
+    tests_fspath = pkg_root.joinpath(*output_module_path.split(".")).parent / "tests"
+
+    logging.info("Running generated tests for %s", output_module_path)
+    # Run generated pytests
+    with add_to_sys_path(pkg_root):
+        result = pytest.main([str(tests_fspath)])
+
+    assert result.value == 0