refactor: move concurrency config checks

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

@@ -563,7 +563,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")

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