TEST: Update Gantt chart tests for coverage

Author: shnizzedy

nipype/pipeline/plugins/tests/test_callback.py

@@ -2,10 +2,14 @@
 # vi: set ft=python sts=4 ts=4 sw=4 et:
 """Tests for workflow callbacks
 """
+from datetime import datetime
+from pathlib import Path
 from time import sleep
+import json
 import pytest
 import nipype.interfaces.utility as niu
 import nipype.pipeline.engine as pe
+from nipype.utils.draw_gantt_chart import _convert_string_to_datetime
 
 try:
     import pandas
@@ -71,22 +75,22 @@ def test_callback_exception(tmpdir, plugin, stop_on_first_crash):
 
 @pytest.mark.parametrize("plugin", ["Linear", "MultiProc", "LegacyMultiProc"])
 @pytest.mark.skipif(not has_pandas, reason="Test requires pandas")
-def test_callback_gantt(tmpdir, plugin):
+def test_callback_gantt(tmp_path: Path, plugin: str) -> None:
     import logging
 
     from os import path
 
     from nipype.utils.profiler import log_nodes_cb
     from nipype.utils.draw_gantt_chart import generate_gantt_chart
 
-    log_filename = path.join(tmpdir, "callback.log")
+    log_filename = tmp_path / "callback.log"
     logger = logging.getLogger("callback")
     logger.setLevel(logging.DEBUG)
     handler = logging.FileHandler(log_filename)
     logger.addHandler(handler)
 
     # create workflow
-    wf = pe.Workflow(name="test", base_dir=tmpdir.strpath)
+    wf = pe.Workflow(name="test", base_dir=str(tmp_path))
     f_node = pe.Node(
         niu.Function(function=func, input_names=[], output_names=[]), name="f_node"
     )
@@ -98,7 +102,21 @@ def test_callback_gantt(tmpdir, plugin):
         plugin_args["n_procs"] = 8
     wf.run(plugin=plugin, plugin_args=plugin_args)
 
-    generate_gantt_chart(
-        path.join(tmpdir, "callback.log"), 1 if plugin == "Linear" else 8
-    )
-    assert path.exists(path.join(tmpdir, "callback.log.html"))
+    with open(log_filename, "r") as _f:
+        loglines = _f.readlines()
+
+        # test missing duration
+        first_line = json.loads(loglines[0])
+        if "duration" in first_line:
+            del first_line["duration"]
+        loglines[0] = json.dumps(first_line)
+
+        # test duplicate timestamp warning
+        loglines.append(loglines[-1])
+
+    with open(log_filename, "w") as _f:
+        _f.write("".join(loglines))
+
+    with pytest.warns(Warning):
+        generate_gantt_chart(str(log_filename), 1 if plugin == "Linear" else 8)
+    assert (tmp_path / "callback.log.html").exists()

nipype/utils/draw_gantt_chart.py

@@ -102,15 +102,25 @@ def log_to_dict(logfile):
 
     nodes_list = [json.loads(l) for l in lines]
 
-    def _convert_string_to_datetime(datestring):
-        try:
+    def _convert_string_to_datetime(
+        datestring: str | datetime.datetime,
+    ) -> datetime.datetime:
+        """Convert a date string to a datetime object."""
+        if isinstance(datestring, datetime.datetime):
+            datetime_object = datestring
+        elif isinstance(datestring, str):
+            date_format = (
+                "%Y-%m-%dT%H:%M:%S.%f%z"
+                if "+" in datestring
+                else "%Y-%m-%dT%H:%M:%S.%f"
+            )
             datetime_object: datetime.datetime = datetime.datetime.strptime(
-                datestring, "%Y-%m-%dT%H:%M:%S.%f"
+                datestring, date_format
             )
-            return datetime_object
-        except Exception as _:
-            pass
-        return datestring
+        else:
+            msg = f"{datestring} is not a string or datetime object."
+            raise TypeError(msg)
+        return datetime_object
 
     date_object_node_list: list = list()
     for n in nodes_list: