Add suspend stats related check (Bugfix)

providers/base/bin/suspend_stats.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+# This file is part of Checkbox.
+#
+# Copyright 2025 Canonical Ltd.
+# Written by:
+#   Hanhsuan Lee <[email protected]>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+import argparse
+import sys
+import os
+
+
+class SuspendStats:
+    """
+    This class is used to parse the information under
+    /sys/power/suspend_stats/
+
+    """
+
+    contents = {}
+
+    def __init__(self):
+        suspend_stat_path = "/sys/power/suspend_stats/"
+        self.contents = self.collect_content_under_directory(suspend_stat_path)
+
+    def collect_content_under_directory(self, search_directory: str) -> dict:
+        """
+        Collect all content under specific directory by filename
+
+        :param search_directory: The directory to search through.
+
+        :returns: collected content by filename
+        """
+        content = {}
+        for root, dirs, files in os.walk(search_directory):
+            for file_name in files:
+                file_path = os.path.join(root, file_name)
+                with open(
+                    file_path, "r", encoding="utf-8", errors="ignore"
+                ) as file:
+                    content[file_name] = file.read().splitlines()[0]
+        return content
+
+    def print_all_content(self):
+        """
+        Print all contents under suspend_stats
+
+        """
+        for c in self.contents:
+            print("{}:{}".format(c, self.contents[c]))
+
+    def is_after_suspend(self) -> bool:
+        """
+        The system is under after suspend status or not
+
+        :returns: return Ture while system is under after suspend status
+        """
+        return (
+            self.contents["success"] != "0"
+            and self.contents["failed_prepare"] == "0"
+            and self.contents["failed_suspend"] == "0"
+            and self.contents["failed_resume"] == "0"
+        )
+
+    def is_any_failed(self) -> bool:
+        """
+        Is any failed during suspend
+
+        :returns: return Ture while one failed during suspend
+        """
+        return self.contents["fail"] != "0"
+
+    def parse_args(self, args=sys.argv[1:]):
+        """
+        command line arguments parsing
+
+        :param args: arguments from sys
+        :type args: sys.argv
+        """
+        parser = argparse.ArgumentParser(
+            prog="suspend status validator",
+            description="Get and valid the content"
+            "under /sys/power/suspend_stats/",
+        )
+
+        subparsers = parser.add_subparsers(dest="type")
+        subparsers.required = True
+
+        # Add parser for validating the system is after suspend or not
+        parser_valid = subparsers.add_parser(
+            "valid", help="validating the system is after suspend or not"
+        )
+        parser_valid.add_argument(
+            "-p",
+            "--print",
+            dest="print",
+            action="store_true",
+            help="Print content",
+        )
+        # Add parser for printing last failed device
+        parser_any = subparsers.add_parser(
+            "any",
+            help="Is there any failed during suspend",
+        )
+        parser_any.add_argument(
+            "-p",
+            "--print",
+            dest="print",
+            action="store_true",
+            help="Print content",
+        )
+
+        return parser.parse_args(args)
+
+    def main(self):
+        args = self.parse_args()
+        if args.type == "valid":
+            if args.print:
+                self.print_all_content()
+            if not self.is_after_suspend():
+                raise SystemExit("System is not under after suspend status")
+        elif args.type == "any":
+            if args.print:
+                self.print_all_content()
+            if self.is_any_failed():
+                raise SystemExit(
+                    "There are [{}] failed".format(self.contents["fail"])
+                )
+
+
+if __name__ == "__main__":
+    SuspendStats().main()

providers/base/tests/test_suspend_stats.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+# This file is part of Checkbox.
+#
+# Copyright 2025 Canonical Ltd.
+# Written by:
+#   Hanhsuan Lee <[email protected]>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+
+from unittest.mock import patch, mock_open, MagicMock
+import unittest
+
+from suspend_stats import SuspendStats
+
+
+class TestSuspendStats(unittest.TestCase):
+    @patch("os.walk")
+    @patch("builtins.open", new_callable=mock_open, read_data="1\n")
+    def test_collect_content_under_directory(self, mock_file, mock_os_walk):
+        mock_os_walk.return_value = [
+            (
+                "/sys/power/suspend_stats/",
+                [],
+                ["success", "failed_suspend", "fail", "last_failed_dev"],
+            ),
+        ]
+
+        stats = SuspendStats()
+        expected_content = {
+            "success": "1",
+            "failed_suspend": "1",
+            "fail": "1",
+            "last_failed_dev": "1",
+        }
+
+        self.assertEqual(stats.contents, expected_content)
+
+    def test_is_after_suspend(self):
+        stats = SuspendStats()
+        stats.contents = {
+            "success": "1",
+            "failed_prepare": "0",
+            "failed_suspend": "0",
+            "failed_resume": "0",
+            "fail": "0",
+            "last_failed_dev": "",
+        }
+        self.assertTrue(stats.is_after_suspend())
+
+        stats.contents["failed_prepare"] = "1"
+        self.assertFalse(stats.is_after_suspend())
+
+        stats.contents["failed_prepare"] = "0"
+        stats.contents["failed_suspend"] = "1"
+        self.assertFalse(stats.is_after_suspend())
+
+        stats.contents["failed_suspend"] = "0"
+        stats.contents["failed_resume"] = "1"
+        self.assertFalse(stats.is_after_suspend())
+
+    def test_is_any_failed(self):
+        stats = SuspendStats()
+        stats.contents = {
+            "success": "1",
+            "failed_suspend": "0",
+            "fail": "1",
+            "last_failed_dev": "",
+        }
+        self.assertTrue(stats.is_any_failed())
+
+        stats.contents["fail"] = "0"
+        self.assertFalse(stats.is_any_failed())
+
+    def test_parse_args_valid(self):
+        stats = SuspendStats()
+        args = ["valid", "--print"]
+        rv = stats.parse_args(args)
+
+        self.assertEqual(rv.type, "valid")
+        self.assertTrue(rv.print)
+
+    def test_parse_args_any(self):
+        stats = SuspendStats()
+        args = ["any", "--print"]
+        rv = stats.parse_args(args)
+
+        self.assertEqual(rv.type, "any")
+        self.assertTrue(rv.print)
+
+
+class MainTests(unittest.TestCase):
+    @patch("suspend_stats.SuspendStats.parse_args")
+    @patch("suspend_stats.SuspendStats.is_after_suspend")
+    @patch("suspend_stats.SuspendStats.print_all_content")
+    def test_run_valid_succ(self, mock_print, mock_after, mock_parse_args):
+        args_mock = MagicMock()
+        args_mock.type = "valid"
+        args_mock.print = True
+        mock_parse_args.return_value = args_mock
+        mock_after.return_value = True
+        self.assertEqual(SuspendStats().main(), None)
+
+    @patch("suspend_stats.SuspendStats.parse_args")
+    @patch("suspend_stats.SuspendStats.is_after_suspend")
+    @patch("suspend_stats.SuspendStats.print_all_content")
+    def test_run_valid_fail(self, mock_print, mock_after, mock_parse_args):
+        args_mock = MagicMock()
+        args_mock.type = "valid"
+        args_mock.print = False
+        mock_parse_args.return_value = args_mock
+        mock_after.return_value = False
+        with self.assertRaises(SystemExit):
+            SuspendStats().main()
+
+    @patch("suspend_stats.SuspendStats.parse_args")
+    @patch("suspend_stats.SuspendStats.is_any_failed")
+    @patch("suspend_stats.SuspendStats.print_all_content")
+    def test_run_any_succ(self, mock_print, mock_any, mock_parse_args):
+        args_mock = MagicMock()
+        args_mock.type = "any"
+        args_mock.print = False
+        mock_parse_args.return_value = args_mock
+        mock_any.return_value = False
+        self.assertEqual(SuspendStats().main(), None)
+
+    @patch("suspend_stats.SuspendStats.parse_args")
+    @patch("suspend_stats.SuspendStats.is_any_failed")
+    @patch("suspend_stats.SuspendStats.print_all_content")
+    def test_run_any_fail(self, mock_print, mock_any, mock_parse_args):
+        args_mock = MagicMock()
+        args_mock.type = "any"
+        args_mock.print = True
+        mock_parse_args.return_value = args_mock
+        mock_any.return_value = True
+        with self.assertRaises(SystemExit):
+            SuspendStats().main()
+
+    @patch("suspend_stats.SuspendStats.parse_args")
+    def test_run_nothing(self, mock_parse_args):
+        args_mock = MagicMock()
+        args_mock.type = "Unknown"
+        args_mock.print = False
+        args_mock.raise_exit = False
+        mock_parse_args.return_value = args_mock
+        self.assertEqual(SuspendStats().main(), None)
+
+
+if __name__ == "__main__":
+    unittest.main()

providers/base/units/suspend/suspend.pxu

@@ -1716,3 +1716,28 @@ command:
  [ -e "${PLAINBOX_SESSION_SHARE}"/fwts_oops_results_after_s3.log ] && xz -c "${PLAINBOX_SESSION_SHARE}"/fwts_oops_results_after_s3.log
 _purpose: Attaches the FWTS oops results log to the submission after suspend
 _summary: Attach FWTS oops results log post-suspend.
+
+id: suspend/valid_suspend_status
+plugin: shell
+category_id: com.canonical.plainbox::suspend
+depends: suspend/suspend_advanced_auto
+estimated_duration: 5s
+command: suspend_stats.py valid -p
+summary:
+    Test this machine suspend status is success or not
+description:
+    If the suspend is successed, the count in /sys/power/suspend_stats/success will be non zero
+    and the /sys/power/suspend_stats/failed_suspend will be zero
+
+id: suspend/any_fail_during_suspend
+plugin: shell
+category_id: com.canonical.plainbox::suspend
+depends: suspend/suspend_advanced_auto
+estimated_duration: 5s
+command: suspend_stats.py any -p
+summary:
+    Test all devices in this machine suspend status is failed or not
+description:
+    If the suspend is successed, the count in /sys/power/suspend_stats/success will be non zero
+    and the /sys/power/suspend_stats/fail will be zero. If there is device failed,
+    /sys/power/suspend_stats/last_failed_dev will show the failed device

providers/base/units/suspend/test-plan.pxu

@@ -46,6 +46,8 @@ include:
     suspend/audio_after_suspend_auto                    certification-status=blocker
     suspend/cpu_after_suspend_auto                      certification-status=blocker
     suspend/memory_after_suspend_auto                   certification-status=blocker
+    suspend/valid_suspend_status
+    suspend/any_fail_during_suspend
 bootstrap_include:
     device