diff --git a/coverage/cmdline.py b/coverage/cmdline.py index 998e6b398..4cd4542b4 100644 --- a/coverage/cmdline.py +++ b/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 diff --git a/doc/cmd.rst b/doc/cmd.rst index ee40a4fe5..791aee1cc 100644 --- a/doc/cmd.rst +++ b/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 ` 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: diff --git a/tests/coveragetest.py b/tests/coveragetest.py index 6b19f1e7d..e9734d633 100644 --- a/tests/coveragetest.py +++ b/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]: diff --git a/tests/helpers.py b/tests/helpers.py index 0861eb854..74c88d861 100644 --- a/tests/helpers.py +++ b/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, diff --git a/tests/test_cmdline.py b/tests/test_cmdline.py index 0ebf27267..37d9a0688 100644 --- a/tests/test_cmdline.py +++ b/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.""" diff --git a/tests/test_process.py b/tests/test_process.py index 5021e1fc8..f95fd259a 100644 --- a/tests/test_process.py +++ b/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")