From 0257922f071809ff1e460628accb93451fb7f812 Mon Sep 17 00:00:00 2001
From: Zhongning Li <60045212+tomli380576@users.noreply.github.com>
Date: Mon, 23 Dec 2024 14:43:56 +0800
Subject: [PATCH] test: add unittests, remove timeout decorator

---
 providers/base/bin/reboot_check_test.py       | 51 ++++++++-----------
 .../base/tests/test_reboot_check_test.py      | 30 +++++++++++
 2 files changed, 52 insertions(+), 29 deletions(-)

diff --git a/providers/base/bin/reboot_check_test.py b/providers/base/bin/reboot_check_test.py
index 8eb9684dd1..d3aea829da 100755
--- a/providers/base/bin/reboot_check_test.py
+++ b/providers/base/bin/reboot_check_test.py
@@ -11,7 +11,6 @@
 from checkbox_support.scripts.image_checker import has_desktop_environment
 from datetime import datetime
 import time
-from checkbox_support.helpers.timeout import timeout
 
 
 # Checkbox could run in a snap container, so we need to prepend this root path
@@ -311,34 +310,28 @@ def wait_for_graphical_target(self, max_wait_seconds: int) -> bool:
         :return: whether graphical.target was reached within max_wait_seconds
         """
 
-        @timeout(max_wait_seconds)
-        def check_systemd():
-            print("Checking if DUT has reached graphical.target...")
-            while True:
-                try:
-                    out = sp.run(
-                        [
-                            "systemd-analyze",
-                            "critical-chain",
-                            "graphical.target",
-                            "--no-pager",
-                        ],
-                        stdout=sp.DEVNULL,
-                        stderr=sp.DEVNULL,
-                    )
-                    if out.returncode == 0:
-                        print("Graphical target reached!")
-                        return True
-                    else:
-                        time.sleep(1)
-                except sp.CalledProcessError:
-                    pass
-
-        try:
-            if check_systemd():
-                return True
-        except TimeoutError:
-            return False
+        start = time.time()
+        print("Checking if DUT has reached graphical.target...")
+        while time.time() - start < max_wait_seconds:
+            try:
+                out = sp.run(
+                    [
+                        "systemd-analyze",
+                        "critical-chain",
+                        "graphical.target",
+                        "--no-pager",
+                    ],
+                    stdout=sp.DEVNULL,
+                    stderr=sp.DEVNULL,
+                )
+                if out.returncode == 0:
+                    print("Graphical target reached!")
+                    return True
+                else:
+                    time.sleep(1)
+            except sp.CalledProcessError:
+                pass
+
         return False
 
     def parse_unity_support_output(
diff --git a/providers/base/tests/test_reboot_check_test.py b/providers/base/tests/test_reboot_check_test.py
index 54d0e6476e..e57ce7c715 100644
--- a/providers/base/tests/test_reboot_check_test.py
+++ b/providers/base/tests/test_reboot_check_test.py
@@ -148,6 +148,36 @@ def test_is_hardware_renderer_available_fail(
         tester = RCT.HardwareRendererTester()
         self.assertFalse(tester.is_hardware_renderer_available())
 
+    def test_slow_boot_scenario(self):
+
+        def fake_time(delta: int):
+            # fake a time.time() delta using closure
+            call_idx = ["start"]
+
+            def wrapped():
+                if call_idx[0] == "start":
+                    call_idx[0] = "end"
+                    return 0  # when time.time is initially called
+                else:
+                    return delta  # the "last" time when time.time is called
+
+            return wrapped
+
+        with patch("subprocess.run") as mock_run, patch(
+            "time.sleep"
+        ) as mock_sleep, patch("time.time") as mock_time:
+            mock_run.side_effect = lambda *args, **kwargs: sp.CompletedProcess(
+                [],
+                1,
+                "systemd says it's not ready",
+                "graphical target not reached blah",
+            )
+            mock_sleep.side_effect = do_nothing
+            mock_time.side_effect = fake_time(3)
+            tester = RCT.HardwareRendererTester()
+
+            self.assertFalse(tester.wait_for_graphical_target(2))
+
 
 class InfoDumpTests(unittest.TestCase):
     @classmethod