wip

Author: nedbat

coverage/collector.py

@@ -14,7 +14,6 @@
 from typing import Any, Callable, TypeVar, cast
 
 from coverage import env
-from coverage.config import CoverageConfig
 from coverage.core import Core
 from coverage.data import CoverageData
 from coverage.debug import short_stack
@@ -60,9 +59,6 @@ class Collector:
     # the top, and resumed when they become the top again.
     _collectors: list[Collector] = []
 
-    # The concurrency settings we support here.
-    LIGHT_THREADS = {"greenlet", "eventlet", "gevent"}
-
     def __init__(
         self,
         core: Core,
@@ -112,8 +108,7 @@ def __init__(
         self.file_mapper = file_mapper
         self.branch = branch
         self.warn = warn
-        self.concurrency = concurrency
-        assert isinstance(self.concurrency, list), f"Expected a list: {self.concurrency!r}"
+        assert isinstance(concurrency, list), f"Expected a list: {concurrency!r}"
 
         self.pid = os.getpid()
 
@@ -125,37 +120,27 @@ def __init__(
 
         self.concur_id_func = None
 
-        # We can handle a few concurrency options here, but only one at a time.
-        concurrencies = set(self.concurrency)
-        unknown = concurrencies - CoverageConfig.CONCURRENCY_CHOICES
-        if unknown:
-            show = ", ".join(sorted(unknown))
-            raise ConfigError(f"Unknown concurrency choices: {show}")
-        light_threads = concurrencies & self.LIGHT_THREADS
-        if len(light_threads) > 1:
-            show = ", ".join(sorted(light_threads))
-            raise ConfigError(f"Conflicting concurrency settings: {show}")
         do_threading = False
 
         tried = "nothing"  # to satisfy pylint
         try:
-            if "greenlet" in concurrencies:
+            if "greenlet" in concurrency:
                 tried = "greenlet"
                 import greenlet
 
                 self.concur_id_func = greenlet.getcurrent
-            elif "eventlet" in concurrencies:
+            elif "eventlet" in concurrency:
                 tried = "eventlet"
                 import eventlet.greenthread
 
                 self.concur_id_func = eventlet.greenthread.getcurrent
-            elif "gevent" in concurrencies:
+            elif "gevent" in concurrency:
                 tried = "gevent"
                 import gevent
 
                 self.concur_id_func = gevent.getcurrent
 
-            if "thread" in concurrencies:
+            if "thread" in concurrency:
                 do_threading = True
         except ImportError as ex:
             msg = f"Couldn't trace with concurrency={tried}, the module isn't installed."
@@ -169,7 +154,7 @@ def __init__(
                 ),
             )
 
-        if do_threading or not concurrencies:
+        if do_threading or not concurrency:
             # It's important to import threading only if we need it.  If
             # it's imported early, and the program being measured uses
             # gevent, then gevent's monkey-patching won't work properly.

coverage/config.py

@@ -390,6 +390,9 @@ def copy(self) -> CoverageConfig:
         "multiprocessing",
     }
 
+    # Mutually exclusive concurrency settings.
+    LIGHT_THREADS = {"greenlet", "eventlet", "gevent"}
+
     CONFIG_FILE_OPTIONS = [
         # These are *args for _set_attr_from_config_option:
         #   (attr, where, type_="")
@@ -563,6 +566,17 @@ def post_process(self) -> None:
         if "subprocess" in self.patch:
             self.parallel = True
 
+        # We can handle a few concurrency options here, but only one at a time.
+        concurrencies = set(self.concurrency)
+        unknown = concurrencies - self.CONCURRENCY_CHOICES
+        if unknown:
+            show = ", ".join(sorted(unknown))
+            raise ConfigError(f"Unknown concurrency choices: {show}")
+        light_threads = concurrencies & self.LIGHT_THREADS
+        if len(light_threads) > 1:
+            show = ", ".join(sorted(light_threads))
+            raise ConfigError(f"Conflicting concurrency settings: {show}")
+
     def debug_info(self) -> list[tuple[str, Any]]:
         """Make a list of (name, value) pairs for writing debug info."""
         return human_sorted_items((k, v) for k, v in self.__dict__.items() if not k.startswith("_"))

coverage/control.py

@@ -562,7 +562,7 @@ def load(self) -> None:
     def _init_for_start(self) -> None:
         """Initialization for start()"""
         # Construct the collector.
-        concurrency: list[str] = self.config.concurrency or []
+        concurrency: list[str] = self.config.concurrency
         if "multiprocessing" in concurrency:
             if self.config.config_file is None:
                 raise ConfigError("multiprocessing requires a configuration file")

coverage/core.py

@@ -69,17 +69,20 @@ def __init__(
             reason_no_sysmon = "can't measure branches in this version"
         elif dynamic_contexts:
             reason_no_sysmon = "doesn't yet support dynamic contexts"
+        elif any((bad := c) in config.concurrency for c in ["greenlet", "eventlet", "gevent"]):
+            reason_no_sysmon = f"doesn't support concurrency={bad}"
 
         core_name: str | None = None
         if config.timid:
             core_name = "pytrace"
-
-        if core_name is None:
+        elif core_name is None:
+            # This could still leave core_name as None.
             core_name = config.core
 
         if core_name == "sysmon" and reason_no_sysmon:
-            warn(f"sys.monitoring {reason_no_sysmon}, using default core", slug="no-sysmon")
-            core_name = None
+            raise ConfigError(
+                f"Can't use core=sysmon: sys.monitoring {reason_no_sysmon}", skip_tests=True
+            )
 
         if core_name is None:
             if env.SYSMON_DEFAULT and not reason_no_sysmon:

coverage/exceptions.py

@@ -11,9 +11,26 @@
 class CoverageException(Exception):
     """The base class of all exceptions raised by Coverage.py."""
 
-    def __init__(self, *args: Any, slug: str | None = None) -> None:
+    def __init__(
+        self,
+        *args: Any,
+        slug: str | None = None,
+        skip_tests: bool = False,
+    ) -> None:
+        """Create an exception.
+
+        Args:
+            slug: A short string identifying the exception, will be used for
+                linking to documentation.
+            skip_tests: If True, raising this exception will skip the test it
+                is raised in.  This is used for shutting off large numbers of
+                tests that we know will not succeed because of a configuration
+                mismatch.
+        """
+
         super().__init__(*args)
         self.slug = slug
+        self.skip_tests = skip_tests
 
 
 class ConfigError(CoverageException):

tests/conftest.py

@@ -104,3 +104,15 @@ def create_pth_file_fixture() -> Iterable[None]:
     finally:
         for p in pth_files:
             p.unlink()
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_call() -> Iterable[None]:
+    """Check the exception raised by the test, and skip the test if needed."""
+    try:
+        yield
+    except Exception as e:
+        if getattr(e, "skip_tests", False):
+            pytest.skip(f"Skipping for exception: {e}")
+        else:
+            raise

tests/test_concurrency.py

@@ -182,7 +182,11 @@ def cant_trace_msg(concurrency: str, the_module: ModuleType | None) -> str | Non
         parts.remove("multiprocessing")
         concurrency = ",".join(parts)
 
-    if the_module is None:
+    if testenv.SYS_MON and concurrency:
+        expected_out = (
+            f"Can't use core=sysmon: sys.monitoring doesn't support concurrency={concurrency}\n"
+        )
+    elif the_module is None:
         # We don't even have the underlying module installed, we expect
         # coverage to alert us to this fact.
         expected_out = (
@@ -251,10 +255,16 @@ def try_some_code(
             lines = line_count(code)
             assert line_counts(data)["try_it.py"] == lines
 
+    @pytest.mark.skipif(
+        not testenv.CAN_MEASURE_THREADS, reason="Can't measure threads with this core."
+    )
     def test_threads(self) -> None:
         code = (THREAD + SUM_RANGE_Q + PRINT_SUM_RANGE).format(QLIMIT=self.QLIMIT)
         self.try_some_code(code, "thread", threading)
 
+    @pytest.mark.skipif(
+        not testenv.CAN_MEASURE_THREADS, reason="Can't measure threads with this core."
+    )
     def test_threads_simple_code(self) -> None:
         code = SIMPLE.format(QLIMIT=self.QLIMIT)
         self.try_some_code(code, "thread", threading)
@@ -318,6 +328,9 @@ def do():
         self.try_some_code(BUG_330, "eventlet", eventlet, "0\n")
 
     # Sometimes a test fails due to inherent randomness. Try more times.
+    @pytest.mark.skipif(
+        not testenv.CAN_MEASURE_THREADS, reason="Can't measure threads with this core."
+    )
     @pytest.mark.flaky(max_runs=3)
     def test_threads_with_gevent(self) -> None:
         self.make_file(
@@ -347,13 +360,17 @@ def gwork(q):
         )
         _, out = self.run_command_status("coverage run --concurrency=thread,gevent both.py")
         if gevent is None:
-            assert out == ("Couldn't trace with concurrency=gevent, the module isn't installed.\n")
+            assert "Couldn't trace with concurrency=gevent, the module isn't installed.\n" in out
             pytest.skip("Can't run test without gevent installed.")
         if not testenv.C_TRACER:
-            assert out == (
-                f"Can't support concurrency=gevent with {testenv.REQUESTED_TRACER_CLASS}, "
-                + "only threads are supported.\n"
-            )
+            if testenv.PY_TRACER:
+                assert out == (
+                    "Can't support concurrency=gevent with PyTracer, only threads are supported.\n"
+                )
+            else:
+                assert out == (
+                    "Can't use core=sysmon: sys.monitoring doesn't support concurrency=gevent\n"
+                )
             pytest.skip(f"Can't run gevent with {testenv.REQUESTED_TRACER_CLASS}.")
 
         assert out == "done\n"
@@ -392,7 +409,10 @@ class WithoutConcurrencyModuleTest(CoverageTest):
     def test_missing_module(self, module: str) -> None:
         self.make_file("prog.py", "a = 1")
         sys.modules[module] = None  # type: ignore[assignment]
-        msg = f"Couldn't trace with concurrency={module}, the module isn't installed."
+        if testenv.SYS_MON:
+            msg = rf"Can't use core=sysmon: sys.monitoring doesn't support concurrency={module}"
+        else:
+            msg = rf"Couldn't trace with concurrency={module}, the module isn't installed."
         with pytest.raises(ConfigError, match=msg):
             self.command_line(f"run --concurrency={module} prog.py")
 
@@ -554,6 +574,9 @@ def test_multiprocessing_and_gevent(self, start_method: str) -> None:
             start_method=start_method,
         )
 
+    @pytest.mark.skipif(
+        not testenv.CAN_MEASURE_BRANCHES, reason="Can't measure branches with this core"
+    )
     def test_multiprocessing_with_branching(self, start_method: str) -> None:
         nprocs = 3
         upto = 30

tests/test_config.py

@@ -83,7 +83,7 @@ def test_toml_config_file(self) -> None:
             [tool.somethingelse]
             authors = ["Joe D'Ávila <[email protected]>"]
             [tool.coverage.run]
-            concurrency = ["a", "b"]
+            concurrency = ["thread", "eventlet"]
             timid = true
             data_file = ".hello_kitty.data"
             plugins = ["plugins.a_plugin"]
@@ -99,7 +99,7 @@ def test_toml_config_file(self) -> None:
         cov = coverage.Coverage()
         assert cov.config.timid
         assert not cov.config.branch
-        assert cov.config.concurrency == ["a", "b"]
+        assert cov.config.concurrency == ["thread", "eventlet"]
         assert cov.config.data_file == ".hello_kitty.data"
         assert cov.config.plugins == ["plugins.a_plugin"]
         assert cov.config.precision == 3

tests/test_core.py

@@ -75,16 +75,21 @@ def test_core_request_pytrace(self) -> None:
 
     def test_core_request_sysmon(self) -> None:
         self.set_environ("COVERAGE_CORE", "sysmon")
-        out = self.run_command("coverage run --debug=sys numbers.py")
-        assert out.endswith("123 456\n")
-        core = re_line(r" core:", out).strip()
-        warns = re_lines(r"\(no-sysmon\)", out)
         if env.PYBEHAVIOR.pep669:
+            status = 0
+        else:
+            status = 1
+        out = self.run_command("coverage run --debug=sys numbers.py", status=status)
+        if status == 0:
+            assert out.endswith("123 456\n")
+            core = re_line(r" core:", out).strip()
+            warns = re_lines(r"\(no-sysmon\)", out)
             assert core == "core: SysMonitor"
             assert not warns
         else:
-            assert core in ["core: CTracer", "core: PyTracer"]
-            assert warns
+            assert out.endswith(
+                "Can't use core=sysmon: sys.monitoring isn't available in this version\n"
+            )
 
     def test_core_request_sysmon_no_dyncontext(self) -> None:
         # Use config core= for this test just to be different.
@@ -96,19 +101,14 @@ def test_core_request_sysmon_no_dyncontext(self) -> None:
             dynamic_context = test_function
             """,
         )
-        out = self.run_command("coverage run --debug=sys numbers.py")
-        assert out.endswith("123 456\n")
-        core = re_line(r" core:", out).strip()
-        assert core in ["core: CTracer", "core: PyTracer"]
-        warns = re_lines(r"\(no-sysmon\)", out)
-        assert len(warns) == 1
+        out = self.run_command("coverage run --debug=sys numbers.py", status=1)
         if env.PYBEHAVIOR.pep669:
             assert (
-                "sys.monitoring doesn't yet support dynamic contexts, using default core"
-                in warns[0]
+                "Can't use core=sysmon: sys.monitoring doesn't yet support dynamic contexts\n"
+                in out
             )
         else:
-            assert "sys.monitoring isn't available in this version, using default core" in warns[0]
+            assert "Can't use core=sysmon: sys.monitoring isn't available in this version\n" in out
 
     def test_core_request_sysmon_no_branches(self) -> None:
         # Use config core= for this test just to be different.
@@ -120,25 +120,23 @@ def test_core_request_sysmon_no_branches(self) -> None:
             branch = True
             """,
         )
-        out = self.run_command("coverage run --debug=sys numbers.py")
-        assert out.endswith("123 456\n")
-        core = re_line(r" core:", out).strip()
-        warns = re_lines(r"\(no-sysmon\)", out)
         if env.PYBEHAVIOR.branch_right_left:
+            status = 0
+        elif env.PYBEHAVIOR.pep669:
+            status = 1
+            msg = "Can't use core=sysmon: sys.monitoring can't measure branches in this version\n"
+        else:
+            status = 1
+            msg = "Can't use core=sysmon: sys.monitoring isn't available in this version\n"
+        out = self.run_command("coverage run --debug=sys numbers.py", status=status)
+        if status == 0:
+            assert out.endswith("123 456\n")
+            core = re_line(r" core:", out).strip()
+            warns = re_lines(r"\(no-sysmon\)", out)
             assert core == "core: SysMonitor"
             assert not warns
         else:
-            assert core in ["core: CTracer", "core: PyTracer"]
-            assert len(warns) == 1
-            if env.PYBEHAVIOR.pep669:
-                assert (
-                    "sys.monitoring can't measure branches in this version, using default core"
-                    in warns[0]
-                )
-            else:
-                assert (
-                    "sys.monitoring isn't available in this version, using default core" in warns[0]
-                )
+            assert out.endswith(msg)  # pylint: disable=possibly-used-before-assignment
 
     def test_core_request_nosuchcore(self) -> None:
         # Test the coverage misconfigurations in-process with pytest. Running a

tests/test_process.py

@@ -1516,6 +1516,9 @@ def test_subprocess_in_directories(self) -> None:
         assert line_counts(data)["main.py"] == 6
         assert line_counts(data)["subproc.py"] == 2
 
+    @pytest.mark.skipif(
+        not testenv.CAN_MEASURE_BRANCHES, reason="Can't measure branches with this core"
+    )
     def test_subprocess_gets_nonfile_config(self) -> None:
         # https://github.com/nedbat/coveragepy/issues/2021
         self.make_file(