Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 4 additions & 9 deletions coverage/cmdline.py
Original file line number Diff line number Diff line change
Expand Up @@ -822,7 +822,7 @@ def do_help(

def do_signal_save(self, _signum: int, _frame: types.FrameType | None) -> None:
""" Signal handler to save coverage report """
print("Saving coverage data ...")
print("Saving coverage data ...", flush=True)
self.coverage.save()

def do_run(self, options: optparse.Values, args: list[str]) -> int:
Expand Down Expand Up @@ -871,15 +871,10 @@ def do_run(self, options: optparse.Values, args: list[str]) -> int:

if options.save_signal:
if env.WINDOWS:
show_help("Signals are not supported in Windows environment.")
return ERR
if options.save_signal.upper() == 'USR1':
signal.signal(signal.SIGUSR1, self.do_signal_save)
elif options.save_signal.upper() == 'USR2':
signal.signal(signal.SIGUSR2, self.do_signal_save)
else:
show_help(f"Unsupported signal for save coverage report: {options.save_signal}")
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()
Expand Down
46 changes: 32 additions & 14 deletions tests/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import os
import os.path
import re
import shlex
import shutil
import subprocess
import sys
Expand All @@ -34,10 +35,23 @@
from coverage.types import TArc


def run_command(cmd: str) -> tuple[int, str]:
def _correct_encoding() -> str:
"""Determine the right encoding to use for subprocesses."""
# Type checking trick due to "unreachable" being set
_locale_type_erased: Any = locale

encoding = os.device_encoding(1) or (
_locale_type_erased.getpreferredencoding()
if sys.version_info < (3, 11)
else _locale_type_erased.getencoding()
)
return encoding


def subprocess_popen(cmd: str, shell: bool=True) -> subprocess.Popen[bytes]:
"""Run a command in a subprocess.

Returns the exit status code and the combined stdout and stderr.
Returns the Popen object.

"""
# Subprocesses are expensive, but convenient, and so may be over-used in
Expand All @@ -47,34 +61,38 @@ def run_command(cmd: str) -> tuple[int, str]:
with open(pth, "a", encoding="utf-8") as proctxt:
print(os.getenv("PYTEST_CURRENT_TEST", "unknown"), file=proctxt, flush=True)

# Type checking trick due to "unreachable" being set
_locale_type_erased: Any = locale

encoding = os.device_encoding(1) or (
_locale_type_erased.getpreferredencoding()
if sys.version_info < (3, 11)
else _locale_type_erased.getencoding()
)

# In some strange cases (PyPy3 in a virtualenv!?) the stdout encoding of
# the subprocess is set incorrectly to ascii. Use an environment variable
# to force the encoding to be the same as ours.
sub_env = dict(os.environ)
sub_env['PYTHONIOENCODING'] = encoding
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,
stderr=subprocess.STDOUT,
)
return proc


def run_command(cmd: str) -> tuple[int, str]:
"""Run a command in a subprocess.

Returns the exit status code and the combined stdout and stderr.

"""
proc = subprocess_popen(cmd)
output, _ = proc.communicate()
status = proc.returncode

# Get the output, and canonicalize it to strings with newlines.
output_str = output.decode(encoding).replace("\r", "")
output_str = output.decode(_correct_encoding()).replace("\r", "")
return status, output_str


Expand Down
14 changes: 14 additions & 0 deletions tests/test_cmdline.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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."""
Expand Down
16 changes: 8 additions & 8 deletions tests/test_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -1171,20 +1171,20 @@ def test_missing_line_ending(self) -> None:
# https://github.com/nedbat/coveragepy/issues/293

self.make_file("normal.py", """\
out, err = subprocess.Popen(
[sys.executable, '-c', 'pass'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE).communicate()
out, err = some_module.some_function(
["my data", "-c", "pass"],
arg1=some_module.NAME,
arg2=some_module.OTHER_NAME).function()
""")

parser = self.parse_file("normal.py")
assert parser.statements == {1}

self.make_file("abrupt.py", """\
out, err = subprocess.Popen(
[sys.executable, '-c', 'pass'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE).communicate()""") # no final newline.
out, err = some_module.some_function(
["my data", "-c", "pass"],
arg1=some_module.NAME,
arg2=some_module.OTHER_NAME).function()""") # no final newline.

# Double-check that some test helper wasn't being helpful.
with open("abrupt.py", encoding="utf-8") as f:
Expand Down
31 changes: 30 additions & 1 deletion tests/test_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -458,6 +460,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.
Expand Down