Merge pull request #146 from ngoldbaum/thread-limit-decorator

Add a parallel_threads_limit mark

Author: ngoldbaum

README.md

@@ -93,9 +93,15 @@ those fixtures are shared between threads.
       as errors.
 
 
-- Three corresponding markers:
-    - `pytest.mark.parallel_threads(n)` to mark a single test to run
-        in parallel in `n` threads
+- Four corresponding markers:
+    - `pytest.mark.parallel_threads_limit(n)` to mark a single test to
+        run in a maximum of `n` threads, even if the `--parallel-threads`
+        command-line argument is set to a higher value. This is useful if a
+        test uses resources that should be limited.
+    - `pytest.mark.parallel_threads(n)` to mark a test to always run in `n`
+        threads. Note that this implies that the test will be multi-threaded
+        just because pytest-run-parallel is installed, even if
+        `--parallel-threads` is not passed at the command-line.
     - `pytest.mark.thread_unsafe` to mark a single test to run in a
         single thread. It is equivalent to using
         `pytest.mark.parallel_threads(1)`

src/pytest_run_parallel/plugin.py

@@ -270,7 +270,7 @@ def pytest_itemcollected(self, item):
             raise ValueError("parallel-threads cannot be negative")
 
         if n_workers == 1 and parallel_threads_marker_used:
-            self._mark_test_thread_unsafe(item, "uses the parallel_threads(1) marker")
+            self._mark_test_thread_unsafe(item, "test is marked as single-threaded")
 
         if n_workers > 1:
             thread_unsafe, reason = self._is_thread_unsafe(item)
@@ -418,7 +418,15 @@ def pytest_configure(config):
     config.addinivalue_line(
         "markers",
         "parallel_threads(n): run the given test function in parallel "
-        "using `n` threads.",
+        "using `n` threads. Note that if n is greater than 1, the test "
+        "run with this many threads even if the --parallel-threads "
+        "command-line argument is not passed. Use parallel_threads_limit "
+        "instead if you want to avoid this pitfall.",
+    )
+    config.addinivalue_line(
+        "markers",
+        "parallel_threads_limit(n): run the given test function in parallel "
+        "using a maximum of `n` threads.",
     )
     config.addinivalue_line(
         "markers",

src/pytest_run_parallel/utils.py

@@ -11,18 +11,30 @@ def get_configured_num_workers(config):
     return n_workers
 
 
+def auto_or_int(val):
+    if val == "auto":
+        logical_cpus = get_logical_cpus()
+        return logical_cpus if logical_cpus is not None else 1
+    return int(val)
+
+
 def get_num_workers(item):
     n_workers = get_configured_num_workers(item.config)
+    # TODO: deprecate in favor of parallel_threads_limit
+    marker_used = False
     marker = item.get_closest_marker("parallel_threads")
     if marker is not None:
-        val = marker.args[0]
-        if val == "auto":
-            logical_cpus = get_logical_cpus()
-            n_workers = logical_cpus if logical_cpus is not None else 1
-        else:
-            n_workers = int(val)
-
-    return n_workers, marker is not None
+        marker_used = True
+        n_workers = auto_or_int(marker.args[0])
+    limit_marker = item.get_closest_marker("parallel_threads_limit")
+    if limit_marker is not None:
+        val = auto_or_int(limit_marker.args[0])
+        if val == 1:
+            marker_used = True
+        if n_workers > val:
+            n_workers = val
+
+    return n_workers, marker_used
 
 
 def get_num_iterations(item):

tests/test_run_parallel.py

@@ -305,6 +305,93 @@ def test_single_threaded(num_parallel_threads):
     )
 
 
+def test_parallel_threads_limit_fixture(pytester):
+    """Test that the num_parallel_threads fixture works as expected."""
+
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+
+        def test_should_yield_global_threads(num_parallel_threads):
+            assert num_parallel_threads == 10
+
+        @pytest.mark.parallel_threads_limit(20)
+        def test_unaffected_by_thread_limit(num_parallel_threads):
+            assert num_parallel_threads == 10
+
+        @pytest.mark.parallel_threads_limit(5)
+        def test_less_than_thread_limit(num_parallel_threads):
+            assert num_parallel_threads == 5
+
+        @pytest.mark.parallel_threads_limit(1)
+        def test_single_threaded(num_parallel_threads):
+            assert num_parallel_threads == 1
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest("--parallel-threads=10", "-v")
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines(
+        [
+            "*::test_unaffected_by_thread_limit PARALLEL PASSED*",
+            "*::test_less_than_thread_limit PARALLEL PASSED*",
+            "*::test_single_threaded PASSED*",
+            "*1 test was not run in parallel because of use of "
+            "thread-unsafe functionality, to list the tests that "
+            "were not run in parallel, re-run while setting PYTEST_RUN_PARALLEL_VERBOSE=1"
+            " in your shell environment",
+        ]
+    )
+
+    # Re-run with verbose output
+    orig = os.environ.get("PYTEST_RUN_PARALLEL_VERBOSE", "0")
+    os.environ["PYTEST_RUN_PARALLEL_VERBOSE"] = "1"
+
+    result = pytester.runpytest("--parallel-threads=10", "-v")
+    os.environ["PYTEST_RUN_PARALLEL_VERBOSE"] = orig
+
+    result.stdout.fnmatch_lines(
+        ["*pytest-run-parallel report*", "*::test_single_threaded*"],
+        consecutive=True,
+    )
+
+
+def test_parallel_threads_limit_one_thread(pytester):
+    """Test that the num_parallel_threads fixture works as expected."""
+
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+
+        def test_should_yield_global_threads(num_parallel_threads):
+            assert num_parallel_threads == 1
+
+        @pytest.mark.parallel_threads_limit(5)
+        def test_marker_threads_five(num_parallel_threads):
+            assert num_parallel_threads == 1
+
+        @pytest.mark.parallel_threads_limit(2)
+        def test_marker_threads_two(num_parallel_threads):
+            assert num_parallel_threads == 1
+
+        @pytest.mark.parallel_threads_limit(1)
+        def test_marker_threads_one(num_parallel_threads):
+            assert num_parallel_threads == 1
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest("-v")
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines(
+        [
+            "*::test_marker_threads_five PASSED [[]???%[]]",
+            "*::test_marker_threads_two PASSED [[]???%[]]",
+            "*::test_marker_threads_one PASSED [[]thread-unsafe[]]*",
+        ]
+    )
+
+
 def test_iterations_marker_one_thread(pytester):
     # create a temporary pytest test module
     pytester.makepyfile("""