feat: Add custom_allowlist to verify_messages function

Signed-off-by: Kacper Tokarzewski <kacper.tokarzewski@intel.com>

Author: KTokarze

README.md

@@ -24,7 +24,7 @@ print(dmesg_obj.get_os_package_info())
 `get_buffer_size_data(self, driver_name: str, driver_interface_number: str) -> Optional[list]` - responsible to return buffer size for respective drivername and interface number.
 `get_os_package_info(self) -> Union[OSPackageInfo, None]` - responsible to return os package installed which is retrived from dmesg output.
 `get_messages_additional(self, service_name: str = None, lines: int = 1000, expected_return_codes: Iterable = frozenset({0}), additional_greps: Optional[List[str]] = None) -> str` - responsible to return latest dmesg output based on addtional filters as specified by the user.
-`verify_messages(self) -> dict` - responsible to check if there are err level messages in dmesg output.
+`verify_messages(self, custom_allowlist: Optional[Iterable[str]] = None) -> dict` - responsible to check if there are err level messages in dmesg output. If `custom_allowlist` is provided, it extends the default `DMESG_WHITELIST` with additional benign error patterns to ignore.
 `clear_messages(self, errors_filter: Optional[List[str]] = [], ignore_filter: Optional[List[str]] = [],) -> Tuple[str, List[str]]` - responsible to clear the message buffer of the kernel (dmesg).
 `clear_messages_after_error(self, error_msg: str) -> Union[Tuple[str, List[str]], None]` - responsible to clear the message buffer of the kernel (dmesg) after user defined error occurred.
 `check_errors(self, error_list: list) -> tuple` - responsible to check for the errors as specified by the user list or user can select from predefined list declared in constant file.

mfd_dmesg/base.py

@@ -220,20 +220,30 @@ def _check_specific_errors(self, error_msg: str) -> bool:
         else:
             return True
 
-    def verify_messages(self) -> dict:
+    def verify_messages(self, custom_allowlist: Iterable[str] | None = None) -> dict:
         """Verify if there are err level messages in dmesg output.
 
+        :param custom_allowlist: list of error messages to ignore in the dmesg output, if any.
+                                 By default, it uses DMESG_WHITELIST which contains known benign errors.
+                                 If this parameter is provided, it will be combined with DMESG_WHITELIST
+                                 to create the final allowlist.
         :return: dictionary indicating success or failure and the error messages if present.
         """
+        if custom_allowlist is None:
+            allowlist = DMESG_WHITELIST
+        else:
+            allowlist = list(DMESG_WHITELIST) + list(custom_allowlist)
         logger.log(level=log_levels.MODULE_DEBUG, msg="Verify Dmesg Errors.")
         level = DmesgLevelOptions.ERRORS if self._is_linux() else DmesgLevelOptions.NONE
         out = self.get_messages(level=level)
         dmesg_result = {"successful": True, "error": ""}
         if out:
             for error in out.splitlines():
+                if not error.strip():
+                    continue
                 is_error = True
                 if self._check_specific_errors(error):
-                    for benign_message in DMESG_WHITELIST:
+                    for benign_message in allowlist:
                         if benign_message in error:
                             is_error = False
                             logger.log(

mfd_dmesg/constants.py

@@ -4,7 +4,7 @@
 
 from dataclasses import dataclass
 from typing import Optional
-from .enums import DmesgLevelOptions # noqa: F401
+from .enums import DmesgLevelOptions  # noqa: F401
 
 
 @dataclass

tests/unit/test_mfd_dmesg/test_base.py

@@ -386,6 +386,63 @@ def test_verify_messages_no_errors(self, dmesg):
         )
         assert dmesg.verify_messages()["successful"]
 
+    def test_verify_messages_ignores_empty_lines(self, dmesg):
+        output = (
+            "[    4.923735] SELinux: Runtime disable is not supported\n"
+            "\n"
+            "[  321.327775] CIFS: VFS: unaligned rsize, making it a multiple of 4096 bytes"
+        )
+        custom_allowlist = ["SELinux", "unaligned rsize"]
+        dmesg._connection.execute_command.return_value = ConnectionCompletedProcess(
+            return_code=0, args="command", stdout=output, stderr="stderr"
+        )
+        result = dmesg.verify_messages(custom_allowlist=custom_allowlist)
+        assert result["successful"]
+        assert result["error"] == ""
+
+    @pytest.mark.parametrize(
+        "custom_allowlist,expected_successful,expected_error_contains",
+        [
+            pytest.param(
+                ["Couldn't get size", "MODSIGN", "unexpected notification"],
+                True,
+                [],
+                id="all_errors_allowlisted",
+            ),
+            pytest.param(
+                ["Couldn't get size"],
+                False,
+                ["MODSIGN", "unexpected notification"],
+                id="partial_match",
+            ),
+            pytest.param(
+                [],
+                False,
+                ["Couldn't get size"],
+                id="empty_allowlist",
+            ),
+        ],
+    )
+    def test_verify_messages_with_custom_allowlist(
+        self, dmesg, custom_allowlist, expected_successful, expected_error_contains
+    ):
+        output = dedent(
+            """
+            [    4.660616] Couldn't get size: 0x800000000000000e
+            [    4.694322] MODSIGN: Couldn't get UEFI db list
+            [   33.580364] cdc_ether 1-1.1.2:1.0 enp0s29u1u1u2: CDC: unexpected notification 20!"""
+        )
+        dmesg._connection.execute_command.return_value = ConnectionCompletedProcess(
+            return_code=0, args="command", stdout=output, stderr="stderr"
+        )
+        result = dmesg.verify_messages(custom_allowlist=custom_allowlist)
+        assert result["successful"] == expected_successful
+        if expected_successful:
+            assert result["error"] == ""
+        else:
+            for fragment in expected_error_contains:
+                assert fragment in result["error"]
+
     def test_check_errors(self, dmesg):
         output = dedent(
             """
@@ -841,6 +898,43 @@ def test_verify_messages_no_errors(self, dmesg):
         )
         assert dmesg.verify_messages()["successful"]
 
+    @pytest.mark.parametrize(
+        "custom_allowlist,expected_successful,expected_error_contains",
+        [
+            pytest.param(
+                ["Couldn't get size", "MODSIGN"],
+                True,
+                [],
+                id="all_errors_allowlisted",
+            ),
+            pytest.param(
+                ["Couldn't get size"],
+                False,
+                ["MODSIGN ERROR"],
+                id="partial_match",
+            ),
+        ],
+    )
+    def test_verify_messages_with_custom_allowlist(
+        self, dmesg, custom_allowlist, expected_successful, expected_error_contains
+    ):
+        output = dedent(
+            """
+            [    4.660616] Couldn't get size error: 0x800000000000000e
+            [    4.694322] MODSIGN ERROR: Couldn't get UEFI db list
+            [    4.728454] Couldn't get size ERROR: 0x800000000000000e"""
+        )
+        dmesg._connection.execute_command.return_value = ConnectionCompletedProcess(
+            return_code=0, args="command", stdout=output, stderr="stderr"
+        )
+        result = dmesg.verify_messages(custom_allowlist=custom_allowlist)
+        assert result["successful"] == expected_successful
+        if expected_successful:
+            assert result["error"] == ""
+        else:
+            for fragment in expected_error_contains:
+                assert fragment in result["error"]
+
     def test_check_errors(self, dmesg):
         output = dedent(
             """