From 1952e15b0d8b4dbff61f290f311c9f696d7aeec6 Mon Sep 17 00:00:00 2001 From: CanYing0913 Date: Thu, 16 Jul 2026 14:38:42 +0800 Subject: [PATCH] Move selection before repeat --- CHANGES.rst | 7 +++ README.rst | 5 ++ pytest_repeat.py | 151 +++++++++++++++++++++++++++++++++++------------ test_repeat.py | 98 ++++++++++++++++++++++++++++++ 4 files changed, 224 insertions(+), 37 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index 1a3e9ddb2..4db3dde5e 100644 --- a/CHANGES.rst +++ b/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 diff --git a/README.rst b/README.rst index f89bb8a33..9cb1cfae6 100644 --- a/README.rst +++ b/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: diff --git a/pytest_repeat.py b/pytest_repeat.py index 455872a01..b233a5ea2 100644 --- a/pytest_repeat.py +++ b/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) diff --git a/test_repeat.py b/test_repeat.py index 101d6a7c6..e3686b8b2 100644 --- a/test_repeat.py +++ b/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