Fix: Update the test_suspend_trigger (Bugfix)

 * Include fixes for the unit tests to succeed with the new
   systemctl callbacks.

Author: EstebanVg15

providers/base/tests/test_suspend_trigger.py

@@ -74,6 +74,7 @@ def test_fwts_path_on_i386_with_defaults(
 
 
 @patch("suspend_trigger.fwts_test")
+@patch("suspend_trigger.subprocess.check_output")
 @patch("suspend_trigger.subprocess.check_call")
 @patch("suspend_trigger.platform.machine")
 @patch("os.remove")
@@ -85,12 +86,14 @@ def test_rtcwake_path_success_with_args(
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
         Tests the rtcwake/systemctl path on aarch64 with custom arguments.
         """
         mock_machine.return_value = "aarch64"
+        mock_check_output.return_value = "No jobs listed."
 
         suspend_trigger.main(
             ["--sleep-delay", "25", "--rtc-device", "/dev/my_rtc"]
@@ -108,25 +111,34 @@ def test_rtcwake_path_success_with_args(
             "25",
         ]
         expected_suspend_cmd = ["systemctl", "suspend"]
+        expected_list_cmd = ["systemctl", "list-jobs", "*suspend*"]
+        mock_check_output.assert_called_with(
+            expected_list_cmd,
+            stderr=subprocess.STDOUT,
+            universal_newlines=True,
+        )
         subprocess_calls = [
             call(expected_rtcwake_cmd),
             call(expected_suspend_cmd),
         ]
         mock_check_call.assert_has_calls(subprocess_calls)
         self.assertEqual(mock_check_call.call_count, 2)
+        self.assertEqual(mock_check_output.call_count, 1)
 
     def test_rtcwake_path_with_defaults(
         self,
         mock_exists,
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
         Tests the rtcwake/systemctl path without any argument.
         """
         mock_machine.return_value = "riscv64"
+        mock_check_output.return_value = "No jobs listed."
 
         suspend_trigger.main([])
 
@@ -141,24 +153,64 @@ def test_rtcwake_path_with_defaults(
             "30",
         ]
         expected_suspend_cmd = ["systemctl", "suspend"]
+        expected_list_cmd = ["systemctl", "list-jobs", "*suspend*"]
+        mock_check_output.assert_called_with(
+            expected_list_cmd,
+            stderr=subprocess.STDOUT,
+            universal_newlines=True,
+        )
         subprocess_calls = [
             call(expected_rtcwake_cmd),
             call(expected_suspend_cmd),
         ]
         mock_check_call.assert_has_calls(subprocess_calls)
 
+    def test_list_command_failure(
+        self,
+        mock_exists,
+        mock_remove,
+        mock_machine,
+        mock_check_call,
+        mock_check_output,
+        mock_fwts_test,
+    ):
+        """
+        Tests the case where the systemctl list-jobs *suspend* fails.
+        """
+        mock_machine.return_value = "aarch64"
+        mock_check_output.return_value = "Suspend jobs ongoing, waiting..."
+        # Simulate a command failure
+        error = subprocess.CalledProcessError(
+            returncode=1,
+            cmd="systemctl list-jobs *suspend*",
+            output="Timed out waiting for suspend jobs to finish"
+        )
+        mock_check_output.side_effect = error
+
+        # The script should propagate the exception
+        with self.assertRaises(subprocess.CalledProcessError):
+            suspend_trigger.main([])
+
+        # Verify that only the first 2 commands were attempted
+        self.assertTrue(mock_check_output.called)
+        called_args = mock_check_output.call_args[0][0]
+        self.assertEqual(called_args[0], "systemctl")
+        self.assertEqual(called_args[1], "list-jobs")
+
     def test_rtcwake_command_failure(
         self,
         mock_exists,
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
         Tests the case where the rtcwake command fails.
         """
         mock_machine.return_value = "aarch64"
+        mock_check_output.return_value = "No jobs listed."
         # Simulate a command failure
         error = subprocess.CalledProcessError(
             returncode=1, cmd="rtcwake", output="Error from rtcwake"
@@ -169,7 +221,8 @@ def test_rtcwake_command_failure(
         with self.assertRaises(subprocess.CalledProcessError):
             suspend_trigger.main([])
 
-        # Verify that only the first command (rtcwake) was attempted
+        # Verify that only the first 2 commands were attempted
+        self.assertTrue(mock_check_output.called)
         self.assertTrue(mock_check_call.called)
         self.assertIn("rtcwake", mock_check_call.call_args[0][0])
 
@@ -179,12 +232,14 @@ def test_suspend_command_failure(
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
         Tests the case where the systemctl suspend command fails.
         """
         mock_machine.return_value = "aarch64"
+        mock_check_output.return_value = "No jobs listed."
         suspend_error = subprocess.CalledProcessError(
             returncode=1, cmd="systemctl suspend", output="Error from suspend"
         )
@@ -195,6 +250,7 @@ def test_suspend_command_failure(
             suspend_trigger.main([])
 
         # Verify both commands were attempted
+        self.assertTrue(mock_check_output.called)
         self.assertEqual(mock_check_call.call_count, 2)
         self.assertIn("rtcwake", mock_check_call.call_args_list[0][0][0])
         self.assertIn("systemctl", mock_check_call.call_args_list[1][0][0])
@@ -205,6 +261,7 @@ def test_log_file_removed_if_exists(
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
@@ -214,10 +271,17 @@ def test_log_file_removed_if_exists(
         mock_machine.return_value = "aarch64"
         mock_exists.return_value = True  # Simulate file exists
         mock_check_call.side_effect = [None, None]
+        mock_check_output.return_value = "No jobs listed."
 
         suspend_trigger.main([])
 
         # Verify commands were called
+        expected_list_cmd = ["systemctl", "list-jobs", "*suspend*"]
+        mock_check_output.assert_called_with(
+            expected_list_cmd,
+            stderr=subprocess.STDOUT,
+            universal_newlines=True,
+        )
         expected_rtcwake_cmd = [
             "rtcwake",
             "--verbose",
@@ -246,6 +310,7 @@ def test_log_file_not_removed_if_missing(
         mock_remove,
         mock_machine,
         mock_check_call,
+        mock_check_output,
         mock_fwts_test,
     ):
         """
@@ -254,6 +319,7 @@ def test_log_file_not_removed_if_missing(
         mock_machine.return_value = "aarch64"
         mock_exists.return_value = False  # Simulate file missing
         mock_check_call.side_effect = [None, None]
+        mock_check_output.return_value = "No jobs listed."
 
         suspend_trigger.main([])
         # Verify remove was never called