141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
"""Failure and hotplug scenarios; no real display or sleep required."""
|
|
|
|
from pathlib import Path
|
|
import runpy
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
guard_module = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/vdi-display-guard"))
|
|
Guard = guard_module["Guard"]
|
|
MODE = guard_module["MODE"]
|
|
|
|
|
|
class FakeRandR:
|
|
def __init__(self):
|
|
self.connected = True
|
|
self.defined = True
|
|
self.attached = True
|
|
self.active = "1920x1440"
|
|
self.writes = []
|
|
self.fail_apply = False
|
|
self.query_error = False
|
|
self.keep_wrong_mode = False
|
|
|
|
def __call__(self, *args):
|
|
if args == ("--current",):
|
|
if self.query_error:
|
|
raise RuntimeError("Can't open display :0")
|
|
status = "connected primary " + ("2560x1440+0+0" if self.active == MODE else "1920x1440+0+0")
|
|
if not self.connected:
|
|
status = "disconnected"
|
|
query = "Screen 0: current 1920 x 1440\nHDMI-1 " + status + " (normal)\n"
|
|
if self.connected:
|
|
query += " 1920x1440 75.00" + ("*" if self.active != MODE else "") + "\n"
|
|
if self.attached:
|
|
query += " " + MODE + " 60.00" + ("*" if self.active == MODE else "") + "\n"
|
|
if self.defined and (not self.attached or not self.connected):
|
|
query += " " + MODE + " (0x189) 297.000MHz +HSync -VSync\n"
|
|
query += " h: width 2560 start 2608\n v: height 1440 start 1443\n"
|
|
return query
|
|
self.writes.append(args)
|
|
if args[0] == "--newmode":
|
|
if self.defined:
|
|
raise RuntimeError("BadName: duplicate mode")
|
|
self.defined = True
|
|
elif args[0] == "--addmode":
|
|
assert self.defined
|
|
self.attached = True
|
|
elif args[0] == "--output":
|
|
if self.fail_apply:
|
|
raise RuntimeError("Configure crtc failed")
|
|
assert self.attached
|
|
if not self.keep_wrong_mode:
|
|
self.active = MODE
|
|
return ""
|
|
|
|
|
|
class DisplayGuardTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.xr = FakeRandR()
|
|
self.logs = []
|
|
self.guard = Guard(self.xr, self.logs.append)
|
|
|
|
def tick(self, now):
|
|
with patch("time.monotonic", return_value=now):
|
|
return self.guard.step(now)
|
|
|
|
def test_mode_drift_restores_after_settling(self):
|
|
self.assertFalse(self.tick(0))
|
|
self.assertEqual(self.xr.writes, [])
|
|
self.assertTrue(self.tick(2))
|
|
self.assertEqual(len(self.xr.writes), 1)
|
|
self.assertEqual(self.xr.active, MODE)
|
|
|
|
def test_healthy_display_does_not_modeset_or_repeat_logs(self):
|
|
self.xr.active = MODE
|
|
for now in range(0, 100, 2):
|
|
self.assertTrue(self.tick(now))
|
|
self.assertEqual(self.xr.writes, [])
|
|
self.assertEqual(len(self.logs), 1)
|
|
|
|
def test_disconnected_never_writes_and_reconnect_settles(self):
|
|
self.tick(0)
|
|
self.xr.connected = False
|
|
for now in range(2, 20, 2):
|
|
self.assertFalse(self.tick(now))
|
|
self.assertEqual(self.xr.writes, [])
|
|
self.xr.connected = True
|
|
self.assertFalse(self.tick(20))
|
|
self.assertTrue(self.tick(22))
|
|
|
|
def test_new_x_server_recreates_and_attaches_mode(self):
|
|
self.xr.defined = self.xr.attached = False
|
|
self.tick(0)
|
|
self.assertTrue(self.tick(2))
|
|
self.assertEqual([w[0] for w in self.xr.writes], ["--newmode", "--addmode", "--output"])
|
|
|
|
def test_hotplug_detached_mode_is_not_created_twice(self):
|
|
self.xr.attached = False
|
|
self.tick(0)
|
|
self.assertTrue(self.tick(2))
|
|
self.assertEqual([w[0] for w in self.xr.writes], ["--addmode", "--output"])
|
|
|
|
def test_late_desktop_settings_overwrite_is_corrected_again(self):
|
|
self.tick(0)
|
|
self.assertTrue(self.tick(2))
|
|
self.xr.active = "1920x1440"
|
|
self.assertFalse(self.tick(4))
|
|
self.assertTrue(self.tick(6))
|
|
self.assertEqual(len(self.xr.writes), 2)
|
|
|
|
def test_modeset_failure_backs_off_and_eventually_recovers(self):
|
|
self.xr.fail_apply = True
|
|
self.tick(0)
|
|
self.tick(2) # retry at 4
|
|
self.tick(4) # retry at 8
|
|
self.tick(6) # no write
|
|
self.assertEqual(len(self.xr.writes), 2)
|
|
self.xr.fail_apply = False
|
|
self.assertTrue(self.tick(8))
|
|
self.assertEqual(len(self.xr.writes), 3)
|
|
|
|
def test_apply_success_must_be_verified(self):
|
|
self.xr.keep_wrong_mode = True
|
|
self.tick(0)
|
|
self.assertFalse(self.tick(2))
|
|
self.assertIn("not retained", self.logs[-1])
|
|
|
|
def test_x11_unavailable_does_not_write_and_then_recovers(self):
|
|
self.xr.query_error = True
|
|
for now in range(0, 20, 2):
|
|
self.assertFalse(self.tick(now))
|
|
self.assertEqual(self.xr.writes, [])
|
|
self.assertEqual(len(self.logs), 1)
|
|
self.xr.query_error = False
|
|
self.assertFalse(self.tick(20))
|
|
self.assertTrue(self.tick(22))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|