Adding support for pytest.Item in additional_information

src/pytest_fluent/additional_information.py

@@ -3,9 +3,18 @@
 import inspect
 import typing
 
-INFORMATION_CALLBACKS: typing.Dict[str, typing.List[typing.Callable]] = {}
+import pytest
+
+INFORMATION_CALLBACKS: typing.Dict[str, typing.List[typing.Callable[..., dict]]] = {}
 SESSION_STAGE = "pytest_sessionstart"
 TEST_STAGE = "pytest_runtest_logstart"
+SUPPORTED_STAGES = [
+    "pytest_sessionstart",
+    "pytest_sessionfinish",
+    "pytest_runtest_logstart",
+    "pytest_runtest_logreport",
+    "pytest_runtest_logfinish",
+]
 
 
 def additional_information_callback(stage_name: str):
@@ -14,6 +23,8 @@ def additional_information_callback(stage_name: str):
     Args:
         stage_name (str): Linked stage name.
     """
+    if stage_name not in SUPPORTED_STAGES:
+        raise ValueError(f"Stage name {stage_name} not supported.")
 
     def wrapper(function: typing.Callable):
         set_additional_information_callback(stage_name, function)
@@ -50,18 +61,21 @@ def set_additional_information_callback(
         stage (str): Stage name.
         function (typing.Callable): Callback function.
     """
+    check_allowed_input(function)
     if stage in INFORMATION_CALLBACKS:
         INFORMATION_CALLBACKS[stage].append(function)
     else:
         INFORMATION_CALLBACKS[stage] = [function]
 
 
 def get_additional_information_callback(
+    item: typing.Optional[pytest.Item] = None,
     stage: typing.Optional[str] = None,
 ) -> typing.Dict[str, typing.Any]:
     """Retrieve stage information from callable.
 
     Args:
+        item (typing.Optional[pytest.Item], optional): Current testcase item.
         stage (typing.Optional[str], optional): Stage name. Defaults to None.
 
     Returns:
@@ -74,9 +88,46 @@ def get_additional_information_callback(
     functions = INFORMATION_CALLBACKS.get(stage)
     if functions is None:
         return info
-    for function in functions:
-        sub_info = function()
-        if not isinstance(sub_info, dict):
-            continue
+    for function in typing.cast(typing.List[typing.Callable[..., dict]], functions):
+        annotations = function.__annotations__
+        if "item" in annotations and annotations["item"] is pytest.Item:
+            sub_info = function(item)
+        else:
+            sub_info = function()
         info.update(sub_info)
     return info
+
+
+def check_type_with_optional(
+    annotation: typing.Any, exptected_type: typing.Type
+) -> bool:
+    """Check if is type with optional.
+
+    Args:
+        annotation (typing.Any): Annotation to check
+        exptected_type (typing.Type): Expected type
+
+    Returns:
+        bool: True if it is the expected type.
+    """
+    is_t = annotation is exptected_type
+    is_opt_t = annotation is typing.Optional[exptected_type]
+    return is_t or is_opt_t
+
+
+def check_allowed_input(func: typing.Callable) -> None:
+    """Check that the given function has a specific signature.
+
+    Args:
+        func (typing.Callable): The function to check.
+    """
+    if not callable(func):
+        raise TypeError("Not a function")
+    annotations = func.__annotations__
+    args = set(annotations.keys())
+    if "item" in args and not check_type_with_optional(
+        annotations["item"], pytest.Item
+    ):
+        raise TypeError("Invalid function signature for 'item'")
+    if not ("return" in args and annotations["return"] is dict):
+        raise TypeError("Invalid function signature for return type. Expecting a dict.")

src/pytest_fluent/plugin.py

@@ -53,6 +53,7 @@ def __init__(self, config):
         self._timestamp = config.getoption("--fluentd-timestamp")
         self._extend_logging = config.getoption("--extend-logging")
         self._add_docstrings = config.getoption("--add-docstrings")
+        self.item: typing.Optional[pytest.Item] = None
         stage_names = [method for method in dir(self) if method.startswith("pytest_")]
         stage_names.append("logging")
         self._content_patcher = ContentPatcher(
@@ -136,6 +137,10 @@ def pytest_sessionstart(self):
             tag, label = self._content_patcher.get_tag_and_label()
             self._event(tag, label, data)
 
+    def pytest_runtest_protocol(self, item: pytest.Item, nextitem: pytest.Item):
+        """Customize hook for a protocol start."""
+        self.item = item
+
     def pytest_runtest_logstart(self, nodeid: str, location: typing.Tuple[int, str]):
         """Customize hook for test start."""
         set_stage("testcase")
@@ -149,7 +154,15 @@ def pytest_runtest_logstart(self, nodeid: str, location: typing.Tuple[int, str])
                 "name": nodeid,
             }
             data = self._content_patcher.patch(data)
-            data.update(get_additional_information_callback())
+            data.update(
+                get_additional_information_callback(
+                    item=(
+                        None
+                        if self.item is None or self.item.nodeid != nodeid
+                        else self.item
+                    )
+                )
+            )
             self._set_timestamp_information(data=data)
             tag, label = self._content_patcher.get_tag_and_label()
             self._event(tag, label, data)
@@ -204,7 +217,15 @@ def pytest_runtest_logreport(self, report: pytest.TestReport):
                     data.update({"docstring": docstring})
             self._set_timestamp_information(data=data)
             data = self._content_patcher.patch(data)
-            data.update(get_additional_information_callback())
+            data.update(
+                get_additional_information_callback(
+                    item=(
+                        None
+                        if self.item is None or self.item.nodeid != report.nodeid
+                        else self.item
+                    )
+                )
+            )
             tag, label = self._content_patcher.get_tag_and_label()
             self._event(tag, label, data)
 

src/pytest_fluent/setting_file_loader_action.py

@@ -38,6 +38,8 @@ def load_and_check_settings_file(file_name: str) -> typing.Dict[str, typing.Any]
         maxsplit=1,
         flags=re.IGNORECASE | re.MULTILINE,
     )
+    file_data = ""
+    data_format = None
     if len(splitted) == 1:
         file_data = splitted[0]
         data_format = None

tests/test_additional_information.py

@@ -1,11 +1,65 @@
+import typing
+from unittest.mock import patch
+
+import pytest
+
+import pytest_fluent.additional_information
 from pytest_fluent import (
     additional_information_callback,
     additional_session_information_callback,
     additional_test_information_callback,
 )
+from pytest_fluent.additional_information import check_allowed_input
+
+
+def test_allowed_input():
+
+    def add_1() -> dict:
+        return {}
+
+    def add_2(item: pytest.Item) -> dict:
+        return {}
+
+    def add_2_opt(item: typing.Optional[pytest.Item] = None) -> dict:
+        return {}
+
+    def add_3(item: int) -> dict:
+        return {}
+
+    def add_4() -> int:
+        return 1
+
+    check_allowed_input(add_1)
+    check_allowed_input(add_2)
+    check_allowed_input(add_2_opt)
+
+    with pytest.raises(TypeError, match="Invalid function signature for 'item'"):
+        check_allowed_input(add_3)
+
+    with pytest.raises(
+        TypeError, match="Invalid function signature for return type. Expecting a dict."
+    ):
+        check_allowed_input(add_4)
 
+    with pytest.raises(TypeError, match="Not a function"):
+        check_allowed_input(1)  # type: ignore
 
-def test_additional_information_convenience_wrapper(run_mocked_pytest, session_uuid):
+
+@patch.object(pytest_fluent.additional_information, "INFORMATION_CALLBACKS", new={})
+def test_additional_information_not_supported(
+    run_mocked_pytest, session_uuid, logging_content
+):
+    with pytest.raises(ValueError):
+
+        @additional_information_callback("test")
+        def test_info() -> dict:
+            return {"type": "myCustomSession"}
+
+
+@patch.object(pytest_fluent.additional_information, "INFORMATION_CALLBACKS", new={})
+def test_additional_information_convenience_wrapper(
+    run_mocked_pytest, session_uuid, logging_content
+):
     @additional_session_information_callback
     def session_info() -> dict:
         return {"type": "myCustomSession"}
@@ -14,17 +68,49 @@ def session_info() -> dict:
     def test_info() -> dict:
         return {"type": "mySuperTestcase"}
 
+    @additional_test_information_callback
+    def test_info_more(item: pytest.Item) -> dict:
+        if not isinstance(item, pytest.Item):
+            return {}
+        marker = [mark for mark in item.own_markers if mark.name == "testcase_id"]
+        testcase_id = None
+        if len(marker) > 0:
+            mark = marker[0]
+            if "id" in mark.kwargs:
+                testcase_id = mark.kwargs["id"]
+            elif len(mark.args) > 0:
+                testcase_id = mark.args[0]
+
+        return {"testcase": {"name": item.nodeid, "id": testcase_id}}
+
     runpytest, fluent_sender = run_mocked_pytest
-    runpytest(f"--session-uuid={session_uuid}")
+    runpytest(
+        f"--session-uuid={session_uuid}",
+        pyfile=f"""
+    from logging import getLogger
+    import pytest
+    LOGGER = getLogger('fluent')
+
+    @pytest.mark.testcase_id(id="1a2b06a4-d902-4d63-9ae1-c16a63e7a9b8")
+    def test_base():
+        LOGGER.info('{logging_content}')
+        assert True
+    """,
+    )
     call_args = fluent_sender.emit_with_time.call_args_list
     for idx, call_arg in enumerate(call_args):
         data = call_arg.args[2]
         if idx == 0:
             assert data.get("type") == "myCustomSession"
         if idx == 1:
             assert data.get("type") == "mySuperTestcase"
+            assert data.get("testcase") == {
+                "name": "test_additional_information_convenience_wrapper.py::test_base",
+                "id": "1a2b06a4-d902-4d63-9ae1-c16a63e7a9b8",
+            }
 
 
+@patch.object(pytest_fluent.additional_information, "INFORMATION_CALLBACKS", new={})
 def test_addtional_information_callbacks(run_mocked_pytest, session_uuid):
     @additional_information_callback("pytest_sessionstart")
     def session_info() -> dict:
@@ -45,6 +131,7 @@ def test_info() -> dict:
             assert data.get("type") == "mySuperTestcase"
 
 
+@patch.object(pytest_fluent.additional_information, "INFORMATION_CALLBACKS", new={})
 def test_multiple_addtional_information_callbacks_per_stage(
     run_mocked_pytest, session_uuid
 ):

tests/test_content_patcher.py

@@ -16,7 +16,9 @@
 @pytest.fixture
 def stage_names() -> typing.List[str]:
     names = [
-        method for method in dir(FluentLoggerRuntime) if method.startswith("pytest_")
+        method
+        for method in dir(FluentLoggerRuntime)
+        if method.startswith("pytest_") and method not in ["pytest_runtest_protocol"]
     ]
     names.append("logging")
     return names