fix: update get_messages_additional method to correctly filter number…

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()
 

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(
             """