refactor(CFMTech#46): Refactor how markers are computed.

An item setup from markers is now computed by the `PyTestMonitorMarkerProcessor` which build a `PyTestMonitorItemConfig`
in turned stashed at the item level.

Author: js-dieu

pytest_monitor/markers.py

@@ -0,0 +1,113 @@
+import warnings
+from typing import Any, Dict, List
+
+import pytest
+
+from pytest_monitor.models import PyTestMonitorItemConfig, PyTestMonitorMarker
+
+
+class SkipTestMarker(PyTestMonitorMarker[bool]):
+    def __init__(self) -> None:
+        super().__init__(
+            name="monitor_skip_test",
+            require_item=False,
+            attribute="skip",
+            doc="Mark the decorated test to be executed but not monitored",
+            default_value=False,
+            deprecated=False,
+        )
+
+    def compute_config_from_item(self, _: pytest.Item, __: pytest.Mark) -> bool:
+        return True
+
+    def post_process(self, configuration: Dict[str, bool]) -> Dict[str, bool]:
+        return configuration
+
+
+class SkipTestIfMarker(PyTestMonitorMarker[bool]):
+    def __init__(self) -> None:
+        super().__init__(
+            name="monitor_skip_test_if",
+            require_item=False,
+            attribute="skip",
+            doc="Mark the decorated test to be executed but not monitored",
+            default_value=False,
+            deprecated=False,
+        )
+
+    def compute_config_from_item(self, _: pytest.Item, marker: pytest.Mark) -> bool:
+        return bool(marker.args[0])
+
+    def post_process(self, configuration: Dict[str, bool]) -> Dict[str, bool]:
+        return configuration
+
+
+class MonitorMarker(PyTestMonitorMarker[bool]):
+    def __init__(self) -> None:
+        super().__init__(
+            name="monitor_test",
+            require_item=False,
+            attribute="force",
+            doc="Mark the decorated test to be monitored (default behaviour)."
+            " This can turn handy to whitelist some test when you have disabled"
+            " monitoring on a whole module",
+            default_value=False,
+            deprecated=False,
+        )
+
+    def compute_config_from_item(self, _: pytest.Item, __: pytest.Mark) -> bool:
+        return True
+
+    def post_process(self, configuration: Dict[str, bool]) -> Dict[str, bool]:
+        if configuration[self.attribute]:
+            configuration["monitor_skip_test"] = True
+        return configuration
+
+
+class MonitorIfMarker(PyTestMonitorMarker[bool]):
+    def __init__(self) -> None:
+        super().__init__(
+            name="monitor_test_if",
+            require_item=False,
+            attribute="force",
+            doc="Mark the decorated test to be monitored if and only if condition"
+            " is verified. This can help you in whitelisting tests to be monitored"
+            " depending on some external conditions.",
+            default_value=False,
+            deprecated=False,
+        )
+
+    def compute_config_from_item(self, _: pytest.Item, marker: pytest.Mark) -> bool:
+        return bool(marker.args[0])
+
+    def post_process(self, configuration: Dict[str, bool]) -> Dict[str, bool]:
+        return configuration
+
+
+class PyTestMonitorMarkerProcessor:
+    def __init__(self):
+        self._markers: List[PyTestMonitorMarker] = [
+            SkipTestMarker(),
+            SkipTestIfMarker(),
+            MonitorMarker(),
+            MonitorIfMarker(),
+        ]
+
+    def _defaults(self) -> Dict[str, Any]:
+        return {marker.attribute: marker.default_value for marker in self._markers}
+
+    def __call__(self, item: pytest.Item) -> PyTestMonitorItemConfig:
+        item_monitor_markers: Dict[str, pytest.Mark] = {
+            marker.name: marker for marker in item.iter_markers() if marker and marker.name.startswith("monitor_")
+        }
+        item_unknown_markers = {marker_name for marker_name in item_monitor_markers if marker_name not in self._markers}
+        for marker_name in item_unknown_markers:
+            warnings.warn(f"Nothing known about marker {marker_name}. Marker will not be considered.")
+        config: Dict[str, bool] = self._defaults()
+        for marker in self._markers:
+            if marker.name in item_monitor_markers:
+                if marker.deprecated:
+                    warnings.warn(f"Marker {marker.name} is deprecated. Consider upgrading your tests")
+                config[marker.attribute] = marker.compute_config_from_item(item, item_monitor_markers[marker.name])
+
+        return PyTestMonitorItemConfig(**config)

pytest_monitor/models.py

@@ -1,6 +1,54 @@
+import abc
 from dataclasses import dataclass, field
+from typing import Dict, Generic, TypeVar
+
+import pytest
+
+T = TypeVar("T")
 
 
 @dataclass(frozen=True)
 class PyTestMonitorConfig:
     enabled: bool = field(default=True, init=True)
+
+
+@dataclass(frozen=True)
+class PyTestMonitorItemConfig:
+    skip: bool = field(default=True, init=True)
+    force: bool = field(default=True, init=True)
+
+
+class PyTestMonitorMarker(abc.ABC, Generic[T]):
+    def __init__(
+        self, name: str, require_item: bool, attribute: str, doc: str, default_value: T, deprecated: bool
+    ) -> None:
+        self._name = name
+        self._require_item = require_item
+        self._attribute = attribute
+        self._doc = doc
+        self._deprecated = deprecated
+        self._default_value = default_value
+
+    @abc.abstractmethod
+    def compute_config_from_item(self, item: pytest.Item, marker: pytest.Mark) -> T:
+        """Compute the marker value to associate to the given item"""
+
+    @abc.abstractmethod
+    def post_process(self, configuration: Dict[str, bool]) -> Dict[str, bool]:
+        """Allow to override some data after the marker has been evaluated"""
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def deprecated(self) -> bool:
+        return self._deprecated
+
+    @property
+    def default_value(self) -> T:
+        return self._default_value
+
+    @property
+    def attribute(self) -> str:
+        return self._attribute

pytest_monitor/pytest_monitor.py

@@ -6,25 +6,14 @@
 import memory_profiler
 import pytest
 
-from pytest_monitor.models import PyTestMonitorConfig
+from pytest_monitor.markers import PyTestMonitorMarkerProcessor
+from pytest_monitor.models import PyTestMonitorConfig, PyTestMonitorItemConfig
 from pytest_monitor.session import PyTestMonitorSession
 
-# These dictionaries are used to compute members set on each items.
-# KEY is the marker set on a test function
-# value is a tuple:
-#  expect_args: boolean
-#  internal marker attribute name: str
-#  callable that set member's value
-#  default value
-PYTEST_MONITOR_VALID_MARKERS = {
-    "monitor_skip_test": (False, "monitor_skip_test", lambda x: True, False),
-    "monitor_skip_test_if": (True, "monitor_skip_test", lambda x: bool(x), False),
-    "monitor_test": (False, "monitor_force_test", lambda x: True, False),
-    "monitor_test_if": (True, "monitor_force_test", lambda x: bool(x), False),
-}
-PYTEST_MONITOR_DEPRECATED_MARKERS = {}
-
 config_stash = pytest.StashKey[PyTestMonitorConfig]()
+item_config_stash = pytest.StashKey[PyTestMonitorItemConfig]()
+
+marker_processor = PyTestMonitorMarkerProcessor()
 
 
 def pytest_addoption(parser):
@@ -102,7 +91,7 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "monitor_skip_test: mark test to be executed but not monitored.")
     config.addinivalue_line(
         "markers",
-        "monitor_skip_test_if(cond): mark test to be executed but " "not monitored if cond is verified.",
+        "monitor_skip_test_if(cond): mark test to be executed but not monitored if cond is verified.",
     )
     config.addinivalue_line(
         "markers",
@@ -126,36 +115,8 @@ def pytest_runtest_setup(item: pytest.Item):
     """
     if not item.session.stash[config_stash].enabled:
         return
-    item_markers = {mark.name: mark for mark in item.iter_markers() if mark and mark.name.startswith("monitor_")}
-    mark_to_del = []
-    for set_marker in item_markers.keys():
-        if set_marker not in PYTEST_MONITOR_VALID_MARKERS:
-            warnings.warn("Nothing known about marker {}. Marker will be dropped.".format(set_marker))
-            mark_to_del.append(set_marker)
-        if set_marker in PYTEST_MONITOR_DEPRECATED_MARKERS:
-            warnings.warn(f"Marker {set_marker} is deprecated. Consider upgrading your tests")
-
-    for marker in mark_to_del:
-        del item_markers[marker]
-
-    all_valid_markers = PYTEST_MONITOR_VALID_MARKERS
-    all_valid_markers.update(PYTEST_MONITOR_DEPRECATED_MARKERS)
-    # Setting instantiated markers
-    for marker, _ in item_markers.items():
-        with_args, attr, fun_val, _ = all_valid_markers[marker]
-        attr_val = fun_val(item_markers[marker].args[0]) if with_args else fun_val(None)
-        setattr(item, attr, attr_val)
-
-    # Setting other markers to default values
-    for marker, marker_value in all_valid_markers.items():
-        with_args, attr, _, default = marker_value
-        if not hasattr(item, attr):
-            setattr(item, attr, default)
 
-    # Finalize marker processing by enforcing some marker's value
-    if item.monitor_force_test:
-        # This test has been explicitly flagged as 'to be monitored'.
-        item.monitor_skip_test = False
+    item.stash[item_config_stash] = marker_processor(item)
 
 
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
@@ -294,7 +255,7 @@ def _prf_tracer(request: pytest.FixtureRequest):
         ptimes_a = request.session.pytest_monitor.process.cpu_times()
         yield
         ptimes_b = request.session.pytest_monitor.process.cpu_times()
-        if not request.node.monitor_skip_test and getattr(request.node, "monitor_results", False):
+        if not request.node.stash[item_config_stash].skip and getattr(request.node, "monitor_results", False):
             item_name = request.node.originalname
             item_loc, *_ = request.node.location
             request.session.pytest_monitor.add_test_info(

tests/test_monitor.py

@@ -181,7 +181,7 @@ def test_ok():
     pymon_path = pathlib.Path(str(testdir)) / ".pymon"
     assert pymon_path.exists()
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     result.assert_outcomes(passed=1)
 
     db = sqlite3.connect(str(pymon_path))
@@ -378,6 +378,8 @@ def test_monitor_with_doctest(testdir):
     # create a temporary pytest test module
     testdir.makepyfile(
         '''
+        pytest_monitor_component = "doctest"
+
         def run(a, b):
             """
             >>> run(3, 30)
@@ -390,7 +392,7 @@ def run(a, b):
     # run pytest with the following cmd args
     result = testdir.runpytest("--doctest-modules", "-vv")
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     result.assert_outcomes(passed=1)
     pymon_path = pathlib.Path(str(testdir)) / ".pymon"
     assert pymon_path.exists()
@@ -403,6 +405,6 @@ def run(a, b):
     pymon_path.unlink()
     result = testdir.runpytest("--doctest-modules", "--no-monitor", "-vv")
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     result.assert_outcomes(passed=1)
     assert not pymon_path.exists()