Handle timezones in started times

Author: samirelanduk

nextflow/command.py

@@ -90,7 +90,7 @@ def _run(
     while True:
         time.sleep(sleep)
         execution, diff = get_execution(
-            output_path, log_path, nextflow_command, execution, log_start
+            output_path, log_path, nextflow_command, execution, log_start, timezone
         )
         log_start += diff
         if execution and poll: yield execution
@@ -251,7 +251,7 @@ def wait_for_log_creation(output_path, start):
         time.sleep(0.1)
 
 
-def get_execution(execution_path, log_path, nextflow_command, execution=None, log_start=0):
+def get_execution(execution_path, log_path, nextflow_command, execution=None, log_start=0, timezone=None):
     """Creates an execution object from a location. If you are polling, you can
     pass in the previous execution to update it with new information.
     
@@ -274,7 +274,7 @@ def get_execution(execution_path, log_path, nextflow_command, execution=None, lo
     for process_execution in process_executions.values():
         if not process_execution.finished or not process_execution.started or \
          process_execution.identifier in changed:
-            update_process_execution_from_path(process_execution, execution_path)
+            update_process_execution_from_path(process_execution, execution_path, timezone)
     execution.process_executions = list(process_executions.values())
     return execution, len(log)
 
@@ -380,7 +380,7 @@ def update_process_execution_from_line(process_executions, line):
     return identifier
 
 
-def update_process_execution_from_path(process_execution, execution_path):
+def update_process_execution_from_path(process_execution, execution_path, timezone=None):
     """Some attributes of a process execution need to be obtained from files on
     disk. This function updates the process execution with these values.
     
@@ -392,7 +392,7 @@ def update_process_execution_from_path(process_execution, execution_path):
     process_execution.stdout = get_file_text(os.path.join(full_path, ".command.out"))
     process_execution.stderr = get_file_text(os.path.join(full_path, ".command.err"))
     if not process_execution.started and not process_execution.cached:
-        process_execution.started = get_file_creation_time(os.path.join(full_path, ".command.begin"))
+        process_execution.started = get_file_creation_time(os.path.join(full_path, ".command.begin"), timezone)
     if not process_execution.bash:
         process_execution.bash = get_file_text(os.path.join(full_path, ".command.sh"))
     if process_execution.execution.finished and not process_execution.return_code:

nextflow/io.py

@@ -1,4 +1,5 @@
 import os
+import pytz
 import glob
 from datetime import datetime
 
@@ -14,14 +15,19 @@ def get_file_text(path):
         return ""
 
 
-def get_file_creation_time(path):
+def get_file_creation_time(path, timezone=None):
     """Gets the creation time of a file.
     
     :param str path: the location of the file.
     :rtype: ``datetime.datetime``"""
 
     try:
-        return datetime.fromtimestamp(os.path.getctime(path))
+        dt = datetime.fromtimestamp(os.path.getctime(path))
+        if timezone:
+            tz = pytz.timezone(timezone)
+            dt = dt.astimezone(tz)
+            dt = dt.replace(tzinfo=None)
+        return dt
     except FileNotFoundError:
         return None
 
@@ -43,4 +49,29 @@ def get_process_ids_to_paths(process_ids, execution_path):
             if sub.startswith(process_id):
                 process_ids_to_paths[process_id] = subdirectory
                 break
-    return process_ids_to_paths
+    return process_ids_to_paths
+
+
+
+
+class CustomIO:
+
+    def read(path, mode="r"):
+        pass
+
+
+    def ctime(path):
+        pass
+
+
+    def listdir(path):
+        pass
+
+
+    def abspath(path):
+        pass
+
+
+    def glob(path):
+        pass
+    

tests/unit/test_command.py

@@ -20,7 +20,7 @@ def test_can_run_with_default_values(self, mock_ex, mock_sleep, mock_run, mock_n
             universal_newlines=True, shell=True
         )
         mock_sleep.assert_called_with(1)
-        mock_ex.assert_called_with(os.path.abspath("."), os.path.abspath("."), mock_nc.return_value, None, 0)
+        mock_ex.assert_called_with(os.path.abspath("."), os.path.abspath("."), mock_nc.return_value, None, 0, None)
         self.assertEqual(executions, [execution])
 
 
@@ -46,7 +46,7 @@ def test_can_run_with_custom_values(self, mock_ex, mock_sleep, mock_wait, mock_r
         mock_wait.assert_called_with("/log", datetime(2025, 1, 1))
         mock_sleep.assert_called_with(4)
         self.assertEqual(mock_sleep.call_count, 3)
-        mock_ex.assert_called_with("/out", "/log", mock_nc.return_value, mock_executions[0], 40)
+        mock_ex.assert_called_with("/out", "/log", mock_nc.return_value, mock_executions[0], 40, "UTC")
         self.assertEqual(mock_ex.call_count, 3)
         self.assertEqual(executions, [mock_executions[1]])
 
@@ -67,7 +67,7 @@ def test_can_run_and_poll(self, mock_ex, mock_sleep, mock_run, mock_nc):
         )
         mock_sleep.assert_called_with(1)
         self.assertEqual(mock_sleep.call_count, 3)
-        mock_ex.assert_called_with("/out", "/out", mock_nc.return_value, mock_executions[0], 60)
+        mock_ex.assert_called_with("/out", "/out", mock_nc.return_value, mock_executions[0], 60, None)
         self.assertEqual(mock_ex.call_count, 3)
         self.assertEqual(executions, mock_executions)
 
@@ -84,7 +84,7 @@ def test_can_run_with_custom_runner(self, mock_ex, mock_sleep, mock_run, mock_nc
         self.assertFalse(mock_run.called)
         runner.assert_called_with(mock_nc.return_value)
         mock_sleep.assert_called_with(1)
-        mock_ex.assert_called_with(os.path.abspath("."), os.path.abspath("."), mock_nc.return_value, None, 0)
+        mock_ex.assert_called_with(os.path.abspath("."), os.path.abspath("."), mock_nc.return_value, None, 0, None)
         self.assertEqual(executions, [mock_ex.return_value[0]])
 
 
@@ -324,17 +324,17 @@ def test_can_get_first_execution(self, mock_update, mock_paths, mock_init, mock_
             "cc/dd": "/ex/cc/dd",
             "gg/hh": "/ex/gg/hh",
         }
-        execution, size = get_execution("/ex", "/log", "nf run")
+        execution, size = get_execution("/ex", "/log", "nf run", timezone="UTC")
         self.assertEqual(execution, mock_execution)
         self.assertEqual(size, 3)
         mock_text.assert_called_with(os.path.join("/log", ".nextflow.log"))
         mock_make.assert_called_with("LOG", "/ex", "nf run", None)
         mock_init.assert_called_with("LOG", mock_execution)
         mock_paths.assert_called_with(["cc/dd","gg/hh"], "/ex")
         self.assertEqual([c[0] for c in mock_update.call_args_list], [
-            (process_executions["aa/bb"], "/ex"),
-            (process_executions["ee/ff"], "/ex"),
-            (process_executions["gg/hh"], "/ex"),
+            (process_executions["aa/bb"], "/ex", "UTC"),
+            (process_executions["ee/ff"], "/ex", "UTC"),
+            (process_executions["gg/hh"], "/ex", "UTC"),
         ])
         self.assertEqual(execution.process_executions, [
             process_executions["aa/bb"],
@@ -364,17 +364,17 @@ def test_can_get_subsequent_execution(self, mock_update, mock_paths, mock_init,
             "cc/dd": "/ex/cc/dd",
             "gg/hh": "/ex/gg/hh",
         }
-        execution, size = get_execution("/ex", "/log", "nf run", mock_execution, 4)
+        execution, size = get_execution("/ex", "/log", "nf run", mock_execution, 4, "UTC")
         self.assertEqual(execution, mock_execution)
         self.assertEqual(size, 3)
         mock_text.assert_called_with(os.path.join("/log", ".nextflow.log"))
         mock_make.assert_called_with("LOG", "/ex", "nf run", mock_execution)
         mock_init.assert_called_with("LOG", mock_execution)
         mock_paths.assert_called_with(["cc/dd","gg/hh"], "/ex")
         self.assertEqual([c[0] for c in mock_update.call_args_list], [
-            (process_executions["aa/bb"], "/ex"),
-            (process_executions["ee/ff"], "/ex"),
-            (process_executions["gg/hh"], "/ex"),
+            (process_executions["aa/bb"], "/ex", "UTC"),
+            (process_executions["ee/ff"], "/ex", "UTC"),
+            (process_executions["gg/hh"], "/ex", "UTC"),
         ])
         self.assertEqual(execution.process_executions, [
             process_executions["aa/bb"],
@@ -633,7 +633,7 @@ def test_can_update_values(self, mock_text):
     def test_can_update_values_with_started(self, mock_time, mock_text):
         proc_ex = Mock(stdout=".", stderr=".", bash=".", finished=None, return_code="", path="aa/bb", started=None, execution=Mock(finished=None), cached=False)
         mock_text.side_effect = ["ok", "bad"]
-        update_process_execution_from_path(proc_ex, "/ex")
+        update_process_execution_from_path(proc_ex, "/ex", timezone="UTC")
         self.assertEqual(proc_ex.stdout, "ok")
         self.assertEqual(proc_ex.stderr, "bad")
         self.assertEqual(proc_ex.bash, ".")
@@ -643,7 +643,7 @@ def test_can_update_values_with_started(self, mock_time, mock_text):
             call(os.path.join("/ex", "work", "aa/bb", ".command.out"),),
             call(os.path.join("/ex", "work", "aa/bb", ".command.err"),),
         ])
-        mock_time.assert_called_with(os.path.join("/ex", "work", "aa/bb", ".command.begin"))
+        mock_time.assert_called_with(os.path.join("/ex", "work", "aa/bb", ".command.begin"), "UTC")
 
 
     @patch("nextflow.command.get_file_text")

tests/unit/test_io.py

@@ -28,6 +28,13 @@ def test_can_get_creation_time(self, mock_getctime):
         mock_getctime.assert_called_with("/ex/file.txt")
 
 
+    @patch("os.path.getctime")
+    def test_can_get_creation_time_with_timezone(self, mock_getctime):
+        mock_getctime.return_value = 123456
+        self.assertEqual(get_file_creation_time("/ex/file.txt", timezone="America/New_York"), datetime.fromtimestamp(123456 - 6 * 60 * 60))
+        mock_getctime.assert_called_with("/ex/file.txt")
+    
+
     @patch("os.path.getctime")
     def test_can_handle_no_file(self, mock_getctime):
         mock_getctime.side_effect = FileNotFoundError