Merge pull request #169 from asmacdo/enh-handle-signals

Modify exit code if cmd terminated by signal

src/con_duct/__main__.py

@@ -436,6 +436,10 @@ def write_subreport(self) -> None:
 
     @property
     def execution_summary(self) -> dict[str, Any]:
+        # killed by a signal
+        # https://pubs.opengroup.org/onlinepubs/9799919799/utilities/V3_chap02.html#tag_19_08_02
+        if self.process and self.process.returncode < 0:
+            self.process.returncode = 128 + abs(self.process.returncode)
         # prepare the base, but enrich if we did get process running
         return {
             "exit_code": self.process.returncode if self.process else None,

test/test_execution.py

@@ -2,7 +2,10 @@
 import json
 import os
 from pathlib import Path
-from time import time
+import signal
+import subprocess
+import threading
+from time import sleep, time
 import pytest
 from utils import assert_files
 from con_duct.__main__ import SUFFIXES, Arguments, Outputs, execute
@@ -193,3 +196,29 @@ def test_execute_unknown_command(
     assert execute(args) == 127
     assert f"{cmd}: command not found\n" == capsys.readouterr().err
     assert_expected_files(temp_output_dir, exists=False)
+
+
+def test_signal_exit(temp_output_dir: str) -> None:
+
+    def runner() -> int:
+        args = Arguments.from_argv(
+            ["sleep", "60.74016230000801"],
+            output_prefix=temp_output_dir,
+        )
+        return execute(args)
+
+    thread = threading.Thread(target=runner)
+    thread.start()
+    sleep(0.03)  # make sure the process is started
+    ps_command = "ps aux | grep '[s]leep 60.74016230000801'"  # brackets to not match grep process
+    ps_output = subprocess.check_output(ps_command, shell=True).decode()
+    pid = int(ps_output.split()[1])
+    os.kill(pid, signal.SIGTERM)
+
+    thread.join()
+    # Cannot retrieve the exit code from the thread, it is written to the file
+    with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info:
+        info_data = json.loads(info.read())
+
+    exit_code = info_data["execution_summary"]["exit_code"]
+    assert exit_code == 128 + 15

test/test_report.py

@@ -168,6 +168,7 @@ def test_execution_summary_formatted(
 
     # Test with process
     report.process = mock_popen
+    report.process.returncode = 0
     output = report.execution_summary_formatted
     assert "None" not in output
     assert "unknown" in output