Move selection before repeat

CHANGES.rst

@@ -1,6 +1,13 @@
 Release Notes
 -------------
 
+**Unreleased**
+
+* Repeat tests after collection filtering instead of parametrizing them during
+  collection. This avoids multiplying collection hook work by ``--count``.
+* Preserve unique repeat node IDs and xdist distribution after the new
+  post-collection expansion.
+
 **0.9.4 (2025-Apr-5)**
 
 * Additions

README.rst

@@ -53,6 +53,11 @@ your test, or tests, to be run:
 
 Each test collected by pytest will be run :code:`count` times.
 
+Collection filtering, including :code:`-k` and :code:`-m`, is applied before
+the selected tests are expanded for repetition. Filters therefore match the
+original test IDs rather than generated repeat suffixes such as
+:code:`[1-10]`.
+
 If you want to mark a test in your code to be repeated a number of times, you
 can use the :code:`@pytest.mark.repeat(count)` decorator:
 

pytest_repeat.py

@@ -1,7 +1,9 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at https://www.mozilla.org/en-US/MPL/2.0/.
+import copy
 import warnings
+from collections import OrderedDict
 from unittest import TestCase
 
 import pytest
@@ -30,44 +32,119 @@ def pytest_configure(config):
         'repeat(n): run the given test function `n` times.')
 
 
-class UnexpectedError(Exception):
-    pass
+_REPEAT_FIXTURE = "__pytest_repeat_step_number"
+_REPEAT_STEP = "__pytest_repeat_step_number"
 
 
 @pytest.fixture()
 def __pytest_repeat_step_number(request):
-    marker = request.node.get_closest_marker("repeat")
-    count = marker and marker.args[0] or request.config.option.count
-    if count > 1:
-        try:
-            return request.param
-        except AttributeError:
-            if issubclass(request.cls, TestCase):
-                warnings.warn(
-                    "Repeating unittest class tests not supported")
-            else:
-                raise UnexpectedError(
-                    "This call couldn't work with pytest-repeat. "
-                    "Please consider raising an issue with your usage.")
-
-
-@pytest.hookimpl(trylast=True)
-def pytest_generate_tests(metafunc):
-    count = metafunc.config.option.count
-    m = metafunc.definition.get_closest_marker('repeat')
-    if m is not None:
-        count = int(m.args[0])
-    if count > 1:
-        metafunc.fixturenames.append("__pytest_repeat_step_number")
-
-        def make_progress_id(i, n=count):
-            return f'{i + 1}-{n}'
-
-        scope = metafunc.config.option.repeat_scope
-        metafunc.parametrize(
-            '__pytest_repeat_step_number',
-            range(count),
-            indirect=True,
-            ids=make_progress_id,
-            scope=scope
-        )
+    return getattr(request.node, _REPEAT_STEP, None)
+
+
+def _get_repeat_count(item):
+    marker = item.get_closest_marker('repeat')
+    if marker is not None:
+        return int(marker.args[0])
+    return item.config.option.count
+
+
+def _make_repeat_name(item, step, count):
+    progress_id = f'{step + 1}-{count}'
+    if item.name.endswith(']'):
+        return f'{item.name[:-1]}-{progress_id}]'
+    return f'{item.name}[{progress_id}]'
+
+
+def _copy_item_state(item, clone):
+    clone.own_markers = list(item.own_markers)
+    clone.extra_keyword_matches = set(item.extra_keyword_matches)
+    clone.user_properties = list(item.user_properties)
+    clone._report_sections = list(item._report_sections)
+    clone.fixturenames = list(item.fixturenames)
+    if _REPEAT_FIXTURE not in clone.fixturenames:
+        clone.fixturenames.append(_REPEAT_FIXTURE)
+
+    clone.keywords = copy.copy(item.keywords)
+    if hasattr(clone.keywords, '_markers'):
+        clone.keywords._markers = dict(item.keywords._markers)
+        clone.keywords._markers.pop(item.name, None)
+        clone.keywords._markers[clone.name] = True
+    if hasattr(clone.keywords, 'node'):
+        clone.keywords.node = clone
+    if hasattr(clone.keywords, 'parent'):
+        clone.keywords.parent = clone.parent
+
+    if hasattr(item, 'stash'):
+        clone.stash = copy.copy(item.stash)
+        clone.stash._storage = dict(item.stash._storage)
+        if hasattr(clone, '_store'):
+            clone._store = clone.stash
+
+    clone._initrequest()
+
+
+def _clone_item(item, step, count):
+    clone = copy.copy(item)
+    clone.name = _make_repeat_name(item, step, count)
+    clone._nodeid = f'{item.parent.nodeid}::{clone.name}'
+    _copy_item_state(item, clone)
+    setattr(clone, _REPEAT_STEP, step)
+    return clone
+
+
+def _expand_item(item):
+    count = _get_repeat_count(item)
+    if count <= 1:
+        return [item]
+    if not isinstance(item, pytest.Function):
+        return [item]
+    if item.cls is not None and issubclass(item.cls, TestCase):
+        warnings.warn("Repeating unittest class tests not supported")
+        return [item]
+    return [_clone_item(item, step, count) for step in range(count)]
+
+
+def _scope_key(item, scope):
+    if scope == 'session':
+        return item.session
+    if scope == 'module':
+        return item.getparent(pytest.Module)
+    if scope == 'class':
+        return item.getparent(pytest.Class) or item.getparent(pytest.Module)
+    return item
+
+
+def _expand_scope(items, scope):
+    groups = OrderedDict()
+    for item in items:
+        groups.setdefault(_scope_key(item, scope), []).append(item)
+
+    expanded = []
+    for group in groups.values():
+        repeated = [(item, _expand_item(item)) for item in group]
+        max_count = max((len(repeats) for _, repeats in repeated), default=0)
+        for step in range(max_count):
+            for _, repeats in repeated:
+                if step < len(repeats):
+                    expanded.append(repeats[step])
+    return expanded
+
+
+def _update_terminal_collection_count(config, items):
+    terminal = config.pluginmanager.getplugin('terminalreporter')
+    if terminal is None or not hasattr(terminal, '_numcollected'):
+        return
+    deselected = len(terminal.stats.get('deselected', ()))
+    terminal._numcollected = len(items) + deselected
+
+
+@pytest.hookimpl(hookwrapper=True, tryfirst=True)
+def pytest_collection_modifyitems(session, config, items):
+    yield
+    scope = config.option.repeat_scope
+    if scope == 'function':
+        items[:] = [repeat for item in items for repeat in _expand_item(item)]
+    else:
+        items[:] = _expand_scope(items, scope)
+    session.testscollected = len(items)
+    _update_terminal_collection_count(config, items)

test_repeat.py

@@ -92,6 +92,104 @@ def test_repeat(parametrized_fixture):
         result.stdout.fnmatch_lines(['*6 passed*'])
         assert result.ret == 0
 
+    def test_repeat_after_deselect(self, testdir):
+        testdir.makeconftest("""
+            collected = 0
+
+            def pytest_itemcollected(item):
+                global collected
+                collected += 1
+
+            def pytest_collection_finish(session):
+                final = len(session.items)
+                print(f'ITEMS: collected={collected}, final={final}')
+        """)
+        testdir.makepyfile("""
+            def test_keep():
+                pass
+
+            def test_drop():
+                pass
+        """)
+        result = testdir.runpytest('-q', '--count', '3', '-k', 'keep')
+        result.stdout.fnmatch_lines([
+            '*ITEMS: collected=2, final=3*',
+            '*3 passed, 1 deselected*',
+        ])
+        assert result.ret == 0
+
+    def test_collection_plugins_only_see_original_items(self, testdir):
+        testdir.makeconftest("""
+            def pytest_collection_modifyitems(items):
+                print(f'MODIFY ITEMS: {len(items)}')
+        """)
+        testdir.makepyfile("""
+            def test_one():
+                pass
+
+            def test_two():
+                pass
+        """)
+        result = testdir.runpytest('-q', '--count', '3')
+        result.stdout.fnmatch_lines([
+            '*MODIFY ITEMS: 2*',
+            '*6 passed*',
+        ])
+        assert result.ret == 0
+
+    def test_repeated_items_have_unique_nodeids(self, testdir):
+        testdir.makeconftest("""
+            def pytest_collection_finish(session):
+                nodeids = [item.nodeid for item in session.items]
+                assert len(nodeids) == len(set(nodeids))
+        """)
+        testdir.makepyfile("""
+            def test_repeat():
+                pass
+        """)
+        result = testdir.runpytest('--count', '3', '--collect-only', '-q')
+        result.stdout.fnmatch_lines([
+            '*::test_repeat[[]1-3[]]*',
+            '*::test_repeat[[]2-3[]]*',
+            '*::test_repeat[[]3-3[]]*',
+            '*3 tests collected*',
+        ])
+        assert result.ret == 0
+
+    def test_function_fixture_isolated_per_repeat(self, testdir):
+        testdir.makepyfile("""
+            import pytest
+
+            active = set()
+
+            @pytest.fixture
+            def resource(request):
+                nodeid = request.node.nodeid
+                assert nodeid not in active
+                active.add(nodeid)
+                yield nodeid
+                active.remove(nodeid)
+
+            def test_repeat(resource, request):
+                assert resource == request.node.nodeid
+        """)
+        result = testdir.runpytest('--count', '3')
+        result.stdout.fnmatch_lines(['*3 passed*'])
+        assert result.ret == 0
+
+    def test_xdist(self, testdir):
+        pytest.importorskip('xdist')
+        testdir.makepyfile("""
+            def test_one():
+                pass
+
+            def test_two():
+                pass
+        """)
+        result = testdir.runpytest_subprocess('-n', '2', '--count', '3')
+        result.stdout.fnmatch_lines(['*6 passed*'])
+        assert result.ret == 0
+
     def test_step_number(self, testdir):
         testdir.makepyfile("""
             import pytest