run: Support system signal as a coverage report dump trigger.

coverage/cmdline.py

@@ -10,9 +10,11 @@
 import os
 import os.path
 import shlex
+import signal
 import sys
 import textwrap
 import traceback
+import types
 
 from typing import cast, Any, NoReturn
 
@@ -188,6 +190,16 @@ class Opts:
             "'pyproject.toml' are tried. [env: COVERAGE_RCFILE]"
         ),
     )
+    save_signal = optparse.make_option(
+        '', '--save-signal', action='store', metavar='SAVE_SIGNAL',
+        choices = ['USR1', 'USR2'],
+        help=(
+            "Define a system signal that will trigger coverage report save operation. " +
+            "It is important that target script do not intercept this signal. " +
+            "Currently supported options are: USR1, USR2. " +
+            "This feature does not work on Windows."
+        ),
+    )
     show_contexts = optparse.make_option(
         "--show-contexts", action="store_true",
         help="Show contexts for covered lines.",
@@ -228,7 +240,6 @@ class Opts:
         help="Display version information and exit.",
     )
 
-
 class CoverageOptionParser(optparse.OptionParser):
     """Base OptionParser for coverage.py.
 
@@ -264,6 +275,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             pylib=None,
             quiet=None,
             rcfile=True,
+            save_signal=None,
             show_contexts=None,
             show_missing=None,
             skip_covered=None,
@@ -523,6 +535,7 @@ def get_prog_name(self) -> str:
             Opts.omit,
             Opts.pylib,
             Opts.parallel_mode,
+            Opts.save_signal,
             Opts.source,
             Opts.timid,
             ] + GLOBAL_ARGS,
@@ -807,6 +820,11 @@ def do_help(
 
         return False
 
+    def do_signal_save(self, _signum: int, _frame: types.FrameType | None) -> None:
+        """ Signal handler to save coverage report """
+        print("Saving coverage data ...", flush=True)
+        self.coverage.save()
+
     def do_run(self, options: optparse.Values, args: list[str]) -> int:
         """Implementation of 'coverage run'."""
 
@@ -851,6 +869,13 @@ def do_run(self, options: optparse.Values, args: list[str]) -> int:
         if options.append:
             self.coverage.load()
 
+        if options.save_signal:
+            if env.WINDOWS:
+                show_help("--save-signal is not supported on Windows.")
+                return ERR
+            sig = getattr(signal, f"SIG{options.save_signal}")
+            signal.signal(sig, self.do_signal_save)
+
         # Run the script.
         self.coverage.start()
         code_ran = True

doc/cmd.rst

@@ -133,6 +133,12 @@ There are many options:
       -p, --parallel-mode   Append the machine name, process id and random number
                             to the data file name to simplify collecting data from
                             many processes.
+      --save-signal=SAVE_SIGNAL
+                            Define a system signal that will trigger coverage
+                            report save operation. It is important that target
+                            script do not intercept this signal. Currently
+                            supported options are: USR1, USR2. This feature does
+                            not work on Windows.
       --source=SRC1,SRC2,...
                             A list of directories or importable names of code to
                             measure.
@@ -143,7 +149,7 @@ There are many options:
       --rcfile=RCFILE       Specify configuration file. By default '.coveragerc',
                             'setup.cfg', 'tox.ini', and 'pyproject.toml' are
                             tried. [env: COVERAGE_RCFILE]
-.. [[[end]]] (sum: saD//ido/B)
+.. [[[end]]] (sum: X8Kbvdq2+f)
 
 If you want :ref:`branch coverage <branch>` measurement, use the ``--branch``
 flag.  Otherwise only statement coverage is measured.
@@ -215,6 +221,9 @@ and may change in the future.
 These options can also be set in the :ref:`config_run` section of your
 .coveragerc file.
 
+In case if you are specifying ``--save-signal``, please make sure that
+your target script doesn't intercept this signal. Otherwise the coverage
+reports will not be generated.
 
 .. _cmd_warnings:
 

tests/coveragetest.py

@@ -418,7 +418,11 @@ def run_command(self, cmd: str, *, status: int = 0) -> str:
 
         """
         actual_status, output = self.run_command_status(cmd)
-        assert actual_status == status
+        if actual_status > 128:
+            # Killed by signal, shell returns 128 + signal_num
+            assert actual_status == 128 + (-1 * status)
+        else:
+            assert actual_status == status
         return output
 
     def run_command_status(self, cmd: str) -> tuple[int, str]:

tests/helpers.py

@@ -13,6 +13,7 @@
 import os
 import os.path
 import re
+import shlex
 import shutil
 import subprocess
 import sys
@@ -47,7 +48,7 @@ def _correct_encoding() -> str:
     return encoding
 
 
-def subprocess_popen(cmd: str) -> subprocess.Popen[bytes]:
+def subprocess_popen(cmd: str, shell: bool=True) -> subprocess.Popen[bytes]:
     """Run a command in a subprocess.
 
     Returns the Popen object.
@@ -66,9 +67,12 @@ def subprocess_popen(cmd: str) -> subprocess.Popen[bytes]:
     sub_env = dict(os.environ)
     sub_env["PYTHONIOENCODING"] = _correct_encoding()
 
+    if not shell:
+        cmd = shlex.split(cmd)  # type: ignore[assignment]
+
     proc = subprocess.Popen(
         cmd,
-        shell=True,
+        shell=shell,
         env=sub_env,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,

tests/test_cmdline.py

@@ -20,6 +20,7 @@
 
 import coverage
 import coverage.cmdline
+from coverage import env
 from coverage.control import DEFAULT_DATAFILE
 from coverage.config import CoverageConfig
 from coverage.exceptions import _ExceptionDuringRun
@@ -942,6 +943,19 @@ def test_no_arguments_at_all(self) -> None:
     def test_bad_command(self) -> None:
         self.cmd_help("xyzzy", "Unknown command: 'xyzzy'")
 
+    def test_save_signal_wrong(self) -> None:
+        self.cmd_help(
+            "run --save-signal=XYZ nothing.py",
+            "option --save-signal: invalid choice: 'XYZ' (choose from 'USR1', 'USR2')",
+        )
+
+    @pytest.mark.skipif(not env.WINDOWS, reason="this is a windows-only error")
+    def test_save_signal_windows(self) -> None:
+        self.cmd_help(
+            "run --save-signal=USR1 nothing.py",
+            "--save-signal is not supported on Windows.",
+        )
+
 
 class CmdLineWithFilesTest(BaseCmdLineTest):
     """Test the command line in ways that need temp files."""

tests/test_process.py

@@ -11,9 +11,11 @@
 import os.path
 import platform
 import re
+import signal
 import stat
 import sys
 import textwrap
+import time
 
 from pathlib import Path
 from typing import Any
@@ -27,7 +29,7 @@
 
 from tests import testenv
 from tests.coveragetest import CoverageTest, TESTS_DIR
-from tests.helpers import re_line, re_lines, re_lines_text
+from tests.helpers import re_line, re_lines, re_lines_text, subprocess_popen
 
 
 class ProcessTest(CoverageTest):
@@ -456,6 +458,33 @@ def test_os_exit(self, patch: bool) -> None:
         else:
             assert seen < total_lines
 
+    @pytest.mark.skipif(env.WINDOWS, reason="Windows can't do --save-signal")
+    @pytest.mark.parametrize("send", [False, True])
+    def test_save_signal(self, send: bool) -> None:
+        # PyPy on Ubuntu seems to need more time for things to happen.
+        base_time = 0.75 if (env.PYPY and env.LINUX) else 0.0
+        self.make_file("loop.py", """\
+            import time
+            print("Starting", flush=True)
+            while True:
+                time.sleep(.02)
+            """)
+        proc = subprocess_popen("coverage run --save-signal=USR1 loop.py", shell=False)
+        time.sleep(base_time + .25)
+        if send:
+            proc.send_signal(signal.SIGUSR1)
+            time.sleep(base_time + .25)
+        proc.kill()
+        proc.wait(timeout=base_time + .25)
+        stdout, _ = proc.communicate()
+        assert b"Starting" in stdout
+        if send:
+            self.assert_exists(".coverage")
+            assert b"Saving coverage data" in stdout
+        else:
+            self.assert_doesnt_exist(".coverage")
+            assert b"Saving coverage data" not in stdout
+
     def test_warnings_during_reporting(self) -> None:
         # While fixing issue #224, the warnings were being printed far too
         # often.  Make sure they're not any more.
@@ -687,6 +716,34 @@ def test_module_name(self) -> None:
         out = self.run_command("python -m coverage")
         assert "Use 'coverage help' for help" in out
 
+    @pytest.mark.skipif(env.WINDOWS, reason="This test is not for Windows")
+    def test_save_signal_usr1(self) -> None:
+        test_file = "dummy_hello.py"
+        self.assert_doesnt_exist(".coverage")
+        self.make_file(test_file, """\
+            import os
+            import signal
+
+            print(f"Sending SIGUSR1 to process {os.getpid()}")
+            os.kill(os.getpid(), signal.SIGUSR1)
+            os.kill(os.getpid(), signal.SIGKILL)
+
+            print('Done and goodbye')
+            """)
+        covered_lines = 4
+        self.run_command(f"coverage run --save-signal USR1 {test_file}", status = -signal.SIGKILL)
+        self.assert_exists(".coverage")
+        data = coverage.CoverageData()
+        data.read()
+        assert line_counts(data)[test_file] == covered_lines
+        out = self.run_command("coverage report")
+        assert out == textwrap.dedent("""\
+            Name             Stmts   Miss  Cover
+            ------------------------------------
+            dummy_hello.py       6      2    67%
+            ------------------------------------
+            TOTAL                6      2    67%
+            """)
 
 TRY_EXECFILE = os.path.join(os.path.dirname(__file__), "modules/process_test/try_execfile.py")