From dafbdb6a2589592c24a0fa97dcfcb6e0ce9513a4 Mon Sep 17 00:00:00 2001 From: "Lasota, Adrian" Date: Tue, 26 May 2026 08:13:30 +0200 Subject: [PATCH] fix: update get_messages_additional method to correctly filter number of lines Signed-off-by: Lasota, Adrian --- mfd_dmesg/base.py | 12 +++---- tests/unit/test_mfd_dmesg/test_base.py | 46 ++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 6 deletions(-) diff --git a/mfd_dmesg/base.py b/mfd_dmesg/base.py index d4993ec..90fe977 100644 --- a/mfd_dmesg/base.py +++ b/mfd_dmesg/base.py @@ -178,12 +178,13 @@ def get_os_package_info(self) -> Union[OSPackageInfo, None]: def get_messages_additional( self, - service_name: str = None, + service_name: str | None = None, lines: int = 1000, - expected_return_codes: Iterable = frozenset({0}), - additional_greps: Optional[List[str]] = None, + expected_return_codes: Iterable[int] = frozenset({0}), + additional_greps: list[str] | None = None, ) -> str: - """Read the last lines of message buffer of the kernel (dmesg). + """ + Read the last lines of message buffer of the kernel (dmesg). :param service_name: limits dmesg messages only to provided service :param lines: limit number of lines @@ -195,7 +196,7 @@ def get_messages_additional( if not additional_greps: additional_greps = [] - command = self._tool_exec + command = f"{self._tool_exec} | tail -n {lines}" if service_name is not None: command += f" | grep '{service_name}'" @@ -204,7 +205,6 @@ def get_messages_additional( for additional_grep in additional_greps: grep_content = grep_content + f"\\|{additional_grep}" if grep_content else f"{additional_grep}" command += f" | grep -i '{grep_content}'" - command += f" | tail -n {lines}" out = self._connection.execute_command(command, shell=True, expected_return_codes=expected_return_codes).stdout return out.strip() diff --git a/tests/unit/test_mfd_dmesg/test_base.py b/tests/unit/test_mfd_dmesg/test_base.py index 60f9bdc..49b7700 100644 --- a/tests/unit/test_mfd_dmesg/test_base.py +++ b/tests/unit/test_mfd_dmesg/test_base.py @@ -223,6 +223,30 @@ def test_get_messages_imc_acc_command_level_error_service_name(self, dmesg): 'dmesg | grep -v "Step" | grep -iE "error|fail" | grep \'ix1\'', shell=True, expected_return_codes={0, 1} ) + def test_get_messages_imc_acc_command_level_warning(self, dmesg): + dmesg._connection.execute_command.side_effect = [ + ConnectionCalledProcessError(returncode=0, cmd=""), + ConnectionCompletedProcess(return_code=0, args="command", stdout="", stderr="stderr"), + ] + dmesg.get_messages(level=DmesgLevelOptions.WARNINGS) + dmesg._connection.execute_command.assert_called_with( + 'dmesg | grep -v "Step" | grep -iE "warning" ', + shell=True, + expected_return_codes={0, 1}, + ) + + def test_get_messages_imc_acc_command_level_critical(self, dmesg): + dmesg._connection.execute_command.side_effect = [ + ConnectionCalledProcessError(returncode=0, cmd=""), + ConnectionCompletedProcess(return_code=0, args="command", stdout="", stderr="stderr"), + ] + dmesg.get_messages(level=DmesgLevelOptions.CRITICAL) + dmesg._connection.execute_command.assert_called_with( + 'dmesg | grep -v "Step" | grep crit ', + shell=True, + expected_return_codes={0, 1}, + ) + def test_get_messages_without_level(self, dmesg): output = dedent( """ @@ -267,6 +291,28 @@ def test_get_messages_additional(self, dmesg): ) assert expected == dmesg.get_messages_additional() + def test_get_messages_additional_command_check(self, dmesg): + """ + Test to verify that correct command is executed. + + when getting additional messages with service name and without level filtering. + """ + output = dedent( + """ + [15197896.276724] IPv6: ens785: IPv6 duplicate address 24:1:1::1 + used by aa:bb:cc:dd:ee:ff detected! + [15222488.275756] ice 0000:4e:00.0 ens786: NIC Link is Down + [15222488.308184] ice 0000:4b:00.0 ens785: NIC Link is Down + [15222506.769174] ice 0000:4e:00.0 ens786: A parallel fault was detected.""" + ) + dmesg._connection.execute_command.return_value = ConnectionCompletedProcess( + return_code=0, args="command", stdout=output, stderr="stderr" + ) + dmesg.get_messages_additional(lines=5, service_name="ice") + dmesg._connection.execute_command.assert_called_with( + f"{dmesg._tool_exec} | tail -n 5 | grep 'ice'", shell=True, expected_return_codes={0} + ) + def test_verify_messages(self, dmesg): output = dedent( """