Add support of critical smoke tests with @pytest.mark.smoke marker an…

…d INI option (#33)

This also includes the following breaking changes:
 - Change pytest_smoke_exclude hook to take precedence over any other options
 - Change the hook name pytest_smoke_always_run to pytest_smoke_include

README.md

@@ -8,8 +8,8 @@ versions](https://img.shields.io/pypi/pyversions/pytest-smoke.svg)](https://pypi
 [![pre-commit.ci status](https://results.pre-commit.ci/badge/github/yugokato/pytest-smoke/main.svg)](https://results.pre-commit.ci/latest/github/yugokato/pytest-smoke/main)
 [![Code style ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://docs.astral.sh/ruff/)
 
-A small `pytest` plugin that enables quick smoke testing against a large test suite by limiting the number of tests from 
-each test function (or specified scope) to a value of `N`.  
+A small `pytest` plugin that enables quick smoke testing against a large test suite by limiting the number of tests 
+executed from each test function (or specified scope) to a value of `N`.  
 You can define `N` as either a fixed number or a percentage, allowing you to scale the test execution down to a smaller 
 subset.
 
@@ -23,8 +23,8 @@ pip install pytest-smoke
 
 ## Usage
 
-The plugin provides the following options to limit the amount of tests to run (`N`, default=`1`) and optionally specify 
-the scope at which `N` is applied.    
+The plugin provides the following options to limit the amount of tests to run (`N`, default=`1`) from each scope 
+(`SCOPE`, default=`function`).  
 If provided, the value of `N` can be either a number (e.g., `5`) or a percentage (e.g., `10%`). 
 ```
 $ pytest -h
@@ -46,7 +46,7 @@ Smoke testing:
 ```
 
 > - The `--smoke-scope` option also supports any custom values, as long as they are handled in the hook
-> - You can override the plugin's default value for `N` and `SCOPE` using INI options. See the "INI Options" section below
+> - You can override the plugin's default values for `N` and `SCOPE` using INI options. See the "INI Options" section below
 > - When using the [pytest-xdist](https://pypi.org/project/pytest-xdist/) plugin for parallel testing, you can configure the `pytest-smoke` plugin to replace the default scheduler with a custom distribution algorithm that distributes tests based on the smoke scope
 
 
@@ -229,6 +229,22 @@ tests/test_something.py::test_something3[17] PASSED              [100%]
 > For any of the above examples, you can change the scope of `N` using the `--smoke-scope` option
 
 
+## Markers
+
+### `@pytest.mark.smoke(mustpass=False)`
+Collected tests explicitly marked with `@pytest.mark.smoke` are considered "critical" smoke tests while ones without 
+this marker are considered "regular" smoke tests. Additionally, if the optional `mustpass` keyword argument is set to 
+`True` in the marker, the test is considered a "must-pass" critical smoke test. 
+
+By default, this categorization has no impact on the plugin. However, when the `smoke_marked_tests_as_critical` 
+INI option is set to `true`, the plugin will apply the following behavior:
+- All collected critical tests are automatically included, in addition to the regular tests selected as part of `N`
+- Execute all critical smoke tests first, before any regular smoke tests
+- If any "must-pass" test fails, all subsequent regular smoke tests will be skipped
+
+> This feature assumes that tests will run sequentially. It will not work when running tests in parallel using a plugin like `pytest-xdist`
+
+
 ## Hooks
 
 The plugin provides the following hooks to customize or extend the plugin's capabilities: 
@@ -238,14 +254,15 @@ This hook allows you to implement your own custom scopes for the `--smoke-scope`
 predefined scopes. Items with the same group ID are grouped together and are considered to be in the same scope, 
 at which `N` is applied. Any custom values passed to the  `--smoke-scope` option must be handled in this hook.
 
-### `pytest_smoke_always_run(item, scope)`
-Return `True` for tests that will always be executed regardless of what options are specified. These items will be 
-considered additional tests and will not be counted towards the calculation of `N`.
+### `pytest_smoke_include(item, scope)`
+Return `True` for tests that should be included as "additional" tests. These tests will not be counted towards the 
+calculation of `N`.
 
 ### `pytest_smoke_exclude(item, scope)`
 Return `True` for tests that should not be selected. These items will not be included in the total number of tests to 
 which `N`% is applied. An example use case is to prevent tests that are marked with `skip` and/or `xfail` from being 
-selected.
+selected.  
+Note that this hook takes precedence over any other options provided by the plugin.
 
 
 ## INI Options
@@ -267,3 +284,7 @@ option replaces the default scheduler with a custom distribution algorithm that
 scope. The custom scheduler will be automatically used when the `-n`/`--numprocesses` option is used without a dist 
 option (`--dist` or `-d`).  
 Plugin default: `false`
+
+### `smoke_marked_tests_as_critical`
+Treat tests marked with `@pytest.mark.smoke` as "critical" smoke tests.    
+Plugin default: `false`

src/pytest_smoke/compat.py

@@ -0,0 +1,27 @@
+import sys
+from enum import Enum
+
+import pytest
+
+if sys.version_info < (3, 11):
+
+    class StrEnum(str, Enum):
+        def _generate_next_value_(name, start, count, last_values) -> str:
+            return name.lower()
+
+        def __str__(self) -> str:
+            return str(self.value)
+else:
+    from enum import StrEnum  # noqa: F401
+
+
+if pytest.version_tuple < (7, 4):
+    from collections.abc import Mapping
+    from typing import NamedTuple, Union
+
+    class TestShortLogReport(NamedTuple):
+        category: str
+        letter: str
+        word: Union[str, tuple[str, Mapping[str, bool]]]
+else:
+    from pytest import TestShortLogReport  # noqa: F401

src/pytest_smoke/hooks.py

@@ -1,4 +1,11 @@
-from pytest import Item, hookspec
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from pytest import hookspec
+
+if TYPE_CHECKING:
+    from pytest import Item
 
 
 @hookspec(firstresult=True)
@@ -11,8 +18,8 @@ def pytest_smoke_generate_group_id(item: Item, scope: str):
 
 
 @hookspec(firstresult=True)
-def pytest_smoke_always_run(item: Item, scope: str):
-    """Return True for tests that will always be executed regardless of what options are specified
+def pytest_smoke_include(item: Item, scope: str):
+    """Return True for tests that should be included as "additional" smoke tests
 
     NOTE: These items will not be counted towards the calculation of N
     """
@@ -22,5 +29,7 @@ def pytest_smoke_always_run(item: Item, scope: str):
 def pytest_smoke_exclude(item: Item, scope: str):
     """Return True for tests that should not be selected
 
-    NOTE: These items will not be included in the total number of tests to which N% is applied
+    NOTE:
+        - These items will not be included in the total number of tests to which N% is applied
+        - This hook takes precedence over any other options provided by the plugin
     """

src/pytest_smoke/plugin.py

@@ -1,29 +1,33 @@
+from __future__ import annotations
+
 import os
 import random
 from collections import Counter
-from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Optional, Union
 from uuid import UUID, uuid4
 
 import pytest
-from pytest import Config, Item, Parser, PytestPluginManager, Session
+from pytest import StashKey
 
 from pytest_smoke import smoke
-from pytest_smoke.types import SmokeDefaultN, SmokeEnvVar, SmokeIniOption, SmokeScope
+from pytest_smoke.compat import TestShortLogReport
+from pytest_smoke.types import SmokeCounter, SmokeDefaultN, SmokeEnvVar, SmokeIniOption, SmokeScope
 from pytest_smoke.utils import generate_group_id, get_scope, parse_ini_option, parse_n, parse_scope, scale_down
 
 if smoke.is_xdist_installed:
     from xdist import is_xdist_controller, is_xdist_worker
 
     from pytest_smoke.extensions.xdist import PytestSmokeXdist
 
-
-DEFAULT_N = SmokeDefaultN(1)
+if TYPE_CHECKING:
+    from pytest import Config, Item, Parser, PytestPluginManager, Session, StashKey, TestReport
 
 
-@dataclass
-class SmokeGroupIDCounter:
-    collected: Counter = field(default_factory=Counter)
-    sellected: Counter = field(default_factory=Counter)
+STASH_KEY_SMOKE_COUNTER = StashKey[SmokeCounter]()
+STASH_KEY_SMOKE_IS_CIRITICAL = StashKey[bool]()
+STASH_KEY_SMOKE_IS_MUSTPASS = StashKey[bool]()
+STASH_KEY_SMOKE_SHOULD_SKIP_RESET = StashKey[bool]()
+DEFAULT_N = SmokeDefaultN(1)
 
 
 @pytest.hookimpl(trylast=True)
@@ -102,6 +106,12 @@ def pytest_addoption(parser: Parser):
         help="[pytest-smoke] When using the pytest-xdist plugin for parallel testing, replace the default scheduler "
         "with a custom distribution algorithm that distributes tests based on the smoke scope",
     )
+    parser.addini(
+        SmokeIniOption.SMOKE_MARKED_TESTS_AS_CRITICAL,
+        type="bool",
+        default=False,
+        help="[pytest-smoke] Treat tests marked with @pytest.mark.smoke as 'critical' smoke tests",
+    )
 
 
 @pytest.hookimpl(tryfirst=True)
@@ -116,6 +126,15 @@ def pytest_configure(config: Config):
             "The --smoke-scope option requires one of --smoke, --smoke-last, or --smoke-random to be specified"
         )
 
+    config.addinivalue_line(
+        "markers",
+        "smoke(mustpass=False): [pytest-smoke] When running smoke tests using the pytest-smoke plugin, the marked test "
+        "is considered a 'critical' smoke test. Additionally if the optional mustpass keyword argument is set to True, "
+        "the test is considered a 'must-pass' critical smoke test. When the feature is explicitly enabled via an INI "
+        "option, critical smoke tests are executed first before regular smoke tests. If any 'must-pass' test fails, "
+        "all subsequent regular smoke tests will be skipped",
+    )
+
     if smoke.is_xdist_installed:
         if config.pluginmanager.has_plugin("xdist"):
             # Register the smoke-xdist plugin if -n/--numprocesses option is given.
@@ -149,13 +168,14 @@ def pytest_collection_modifyitems(session: Session, config: Config, items: list[
                 num_smoke = float(n[:-1])
             else:
                 num_smoke = n
+
             scope = get_scope(config)
-            selected_items = []
+            selected_items_regular = []
+            selected_items_critical = []
             deselected_items = []
-            counter = SmokeGroupIDCounter(
-                collected=Counter(filter(None, (generate_group_id(item, scope) for item in items)))
-            )
             smoke_groups_reached_threshold = set()
+            counter = SmokeCounter(collected=Counter(filter(None, (generate_group_id(item, scope) for item in items))))
+            session.stash[STASH_KEY_SMOKE_COUNTER] = counter
             if config.option.smoke_random:
                 if smoke.is_xdist_installed and (is_xdist_controller(session) or is_xdist_worker(session)):
                     # Set the seed to ensure XDIST controler and workers collect the same items
@@ -169,28 +189,92 @@ def pytest_collection_modifyitems(session: Session, config: Config, items: list[
                 items_to_filter = items
 
             for item in items_to_filter:
-                if config.hook.pytest_smoke_always_run(item=item, scope=scope):
-                    # Will not be counted towards the calculation of N
-                    selected_items.append(item)
+                group_id = generate_group_id(item, scope)
+                if group_id is None:
+                    deselected_items.append(item)
                     continue
 
-                group_id = generate_group_id(item, scope)
-                if group_id is None or group_id in smoke_groups_reached_threshold:
+                # Tests that match the below conditions will not be counted towards the calculation of N
+                smoke_marker = item.get_closest_marker("smoke")
+                if smoke_marker and parse_ini_option(config, SmokeIniOption.SMOKE_MARKED_TESTS_AS_CRITICAL):
+                    selected_items_critical.append(item)
+                    if is_mustpass := smoke_marker.kwargs.get("mustpass", False) is True:
+                        counter.mustpass.selected.add(item)
+                    item.stash[STASH_KEY_SMOKE_IS_CIRITICAL] = True
+                    item.stash[STASH_KEY_SMOKE_IS_MUSTPASS] = is_mustpass
+                    continue
+                elif config.hook.pytest_smoke_include(item=item, scope=scope):
+                    selected_items_regular.append(item)
+                    continue
+
+                if group_id in smoke_groups_reached_threshold:
                     deselected_items.append(item)
                     continue
 
                 threshold = scale_down(counter.collected[group_id], num_smoke) if is_scale else num_smoke
-                if counter.sellected[group_id] < threshold:
-                    counter.sellected.update([group_id])
-                    selected_items.append(item)
+                if counter.selected[group_id] < threshold:
+                    counter.selected.update([group_id])
+                    selected_items_regular.append(item)
                 else:
                     smoke_groups_reached_threshold.add(group_id)
                     deselected_items.append(item)
 
-            if deselected_items:
-                config.hook.pytest_deselected(items=deselected_items)
+            assert len(items) == len(selected_items_critical + selected_items_regular + deselected_items)
+            if selected_items_critical or deselected_items:
+                if deselected_items:
+                    config.hook.pytest_deselected(items=deselected_items)
+
                 if config.option.smoke_random or config.option.smoke_last:
                     # retain the original test order
-                    selected_items.sort(key=lambda x: items.index(x))
+                    for smoke_items in (selected_items_critical, selected_items_regular):
+                        if smoke_items:
+                            smoke_items.sort(key=lambda x: items.index(x))
+
                 items.clear()
-                items.extend(selected_items)
+                items.extend(selected_items_critical + selected_items_regular)
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_protocol(item: Item, nextitem: Optional[Item]):
+    try:
+        return (yield)
+    finally:
+        if nextitem and item.stash.get(STASH_KEY_SMOKE_IS_CIRITICAL, False):
+            counter = item.session.stash[STASH_KEY_SMOKE_COUNTER].mustpass
+            if counter.failed and not nextitem.stash.get(STASH_KEY_SMOKE_IS_CIRITICAL, False):
+                # At least one must-pass test failed, and this is the last critical test.
+                # Set the flag to skip all subsequent regular tests
+                item.session.stash[STASH_KEY_SMOKE_SHOULD_SKIP_RESET] = True
+
+
+@pytest.hookimpl(tryfirst=True)
+def pytest_runtest_setup(item: Item):
+    if item.session.stash.get(STASH_KEY_SMOKE_SHOULD_SKIP_RESET, False):
+        counter = item.session.stash[STASH_KEY_SMOKE_COUNTER].mustpass
+        num_selected = len(counter.selected)
+        num_failed = len(counter.failed)
+        pytest.skip(reason=f"{num_failed}/{num_selected} must-pass smoke test{'s' if num_failed > 1 else ''} failed")
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_makereport(item: Item):
+    report: TestReport = yield
+    if item.stash.get(STASH_KEY_SMOKE_IS_MUSTPASS, False):
+        setattr(report, "_is_smoke_must_pass", True)
+        if report.failed:
+            item.session.stash[STASH_KEY_SMOKE_COUNTER].mustpass.failed.add(item)
+    return report
+
+
+@pytest.hookimpl(wrapper=True, trylast=True)
+def pytest_report_teststatus(report: TestReport):
+    status: Union[tuple, TestShortLogReport] = yield
+    if not isinstance(status, TestShortLogReport):
+        status = TestShortLogReport(*status)
+    if status.word and getattr(report, "_is_smoke_must_pass", False):
+        annot = " (must-pass)"
+        if isinstance(status.word, str):
+            status = status._replace(word=status.word + annot)
+        elif isinstance(status.word, tuple):
+            status = status._replace(word=([status.word[0] + annot, *status.word[1:]]))
+    return status

src/pytest_smoke/types.py

@@ -1,16 +1,14 @@
-import sys
-from enum import Enum, auto
+from __future__ import annotations
 
-if sys.version_info < (3, 11):
+from collections import Counter
+from dataclasses import dataclass, field
+from enum import auto
+from typing import TYPE_CHECKING
 
-    class StrEnum(str, Enum):
-        def _generate_next_value_(name, start, count, last_values) -> str:
-            return name.lower()
+if TYPE_CHECKING:
+    from pytest import Item
 
-        def __str__(self) -> str:
-            return str(self.value)
-else:
-    from enum import StrEnum
+from pytest_smoke.compat import StrEnum
 
 
 class SmokeEnvVar(StrEnum):
@@ -29,6 +27,20 @@ class SmokeIniOption(StrEnum):
     SMOKE_DEFAULT_N = auto()
     SMOKE_DEFAULT_SCOPE = auto()
     SMOKE_DEFAULT_XDIST_DIST_BY_SCOPE = auto()
+    SMOKE_MARKED_TESTS_AS_CRITICAL = auto()
 
 
 class SmokeDefaultN(int): ...
+
+
+@dataclass
+class MustpassCounter:
+    selected: set[Item] = field(default_factory=set)
+    failed: set[Item] = field(default_factory=set)
+
+
+@dataclass
+class SmokeCounter:
+    collected: Counter = field(default_factory=Counter)
+    selected: Counter = field(default_factory=Counter)
+    mustpass: MustpassCounter = field(default_factory=MustpassCounter)