test: add unittests, remove timeout decorator

canonical · Dec 23, 2024 · 0257922 · 0257922
1 parent 9dcb9d5
commit 0257922
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 29 deletions.
diff --git 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
@@ -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