TEST: Update Gantt chart tests for coverage

nipype/pipeline/plugins/tests/test_callback.py

@@ -2,7 +2,10 @@
 # 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
@@ -71,22 +74,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 +101,17 @@ 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()
+        # replace 'start' string in first line with datetime to test type handling
+        first_line: dict[str, str | int | float | datetime] = json.loads(loglines[0])
+        assert isinstance(first_line["start"], str)
+        first_line["start"] = datetime.strptime(
+            first_line["start"], "%Y-%m-%dT%H:%M:%S.%f"
+        )
+        loglines[0] = json.dumps(first_line)
+        loglines.append(loglines[-1])  # test duplicate timestamp warning
+
+    with pytest.warns(Warning):
+        generate_gantt_chart(str(log_filename), 1 if plugin == "Linear" else 8)
+    assert (tmp_path / "callback.log.html").exists()