diff --git a/README.md b/README.md index 0c97919..2e008ee 100644 --- a/README.md +++ b/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. diff --git a/mfd_dmesg/base.py b/mfd_dmesg/base.py index d4993ec..abeddf9 100644 --- a/mfd_dmesg/base.py +++ b/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( diff --git a/mfd_dmesg/constants.py b/mfd_dmesg/constants.py index d8dcc15..4692a95 100644 --- a/mfd_dmesg/constants.py +++ b/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 diff --git a/tests/unit/test_mfd_dmesg/test_base.py b/tests/unit/test_mfd_dmesg/test_base.py index 60f9bdc..8e7e047 100644 --- a/tests/unit/test_mfd_dmesg/test_base.py +++ b/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( """