279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""Behavior tests using fake IPC, udev, sudo and systemd; never switch real VTs."""
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MOCK = r'''#!/usr/bin/env python3
|
|
import json, os, pathlib, signal, sys, time
|
|
name = pathlib.Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
state = pathlib.Path(os.environ['TEST_STATE'])
|
|
with (state / 'calls').open('a') as f:
|
|
f.write(json.dumps([name] + args) + '\n')
|
|
if name == 'loginctl':
|
|
if args[0] == 'list-sessions':
|
|
if not os.environ.get('NO_SESSION'):
|
|
print('7 1000 desktop seat0 tty1')
|
|
else:
|
|
active = 'no' if (state / 'inactive').exists() else 'yes'
|
|
print(f"User={os.environ.get('SESSION_UID', os.getuid())}\nType={os.environ.get('SESSION_TYPE', 'wayland')}\nActive={active}\nVTNr={os.environ.get('SESSION_VT', '1')}\nSeat={os.environ.get('SESSION_SEAT', 'seat0')}\nDesktop={os.environ.get('SESSION_DESKTOP', 'Hyprland')}")
|
|
elif name == 'hyprctl':
|
|
if os.environ.get('HYPR_FAIL'):
|
|
sys.exit(1)
|
|
if args == ['-j', 'monitors']:
|
|
if os.environ.get('BAD_JSON'):
|
|
print('invalid')
|
|
else:
|
|
print(json.dumps([{'name': os.environ.get('OUTPUT', 'HDMI-A-2')}]))
|
|
else:
|
|
print('ok')
|
|
elif name == 'sleep':
|
|
if os.environ.get('CHANGE_SESSION'):
|
|
(state / 'inactive').touch()
|
|
if os.environ.get('SIGNAL_AFTER_SWITCH') and (state / 'away').exists():
|
|
os.kill(os.getppid(), signal.SIGTERM)
|
|
if os.environ.get('BLOCK_AFTER_SWITCH') and (state / 'away').exists():
|
|
(state / 'sleeping').touch()
|
|
time.sleep(30)
|
|
elif name == 'sudo':
|
|
vt = args[-1]
|
|
if '-l' in args:
|
|
if os.environ.get('CHANGE_DURING_PERMISSION_CHECK'):
|
|
(state / 'inactive').touch()
|
|
sys.exit(1 if os.environ.get('DENY_RETURN') and vt == '1' else 0)
|
|
if vt == '2':
|
|
(state / 'away').touch()
|
|
if vt == '1' and os.environ.get('RETURN_FAIL_ONCE') and not (state / 'return-tried').exists():
|
|
(state / 'return-tried').touch()
|
|
sys.exit(1)
|
|
if os.environ.get('SUDO_FAIL'):
|
|
sys.exit(1)
|
|
elif name == 'udevadm':
|
|
for _ in range(int(os.environ.get('EVENT_COUNT', '1'))):
|
|
print('KERNEL[1.0] change /devices/drm/card0 (drm)\nHOTPLUG=1\n', flush=True)
|
|
sys.exit(int(os.environ.get('UDEV_EXIT', '0')))
|
|
elif name == 'fake-fix':
|
|
sys.exit(int(os.environ.get('FIX_EXIT', '0')))
|
|
elif name == 'systemctl':
|
|
if 'is-active' in args:
|
|
sys.exit(0 if os.environ.get('SERVICE_ACTIVE') else 3)
|
|
'''
|
|
|
|
|
|
class RecoveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
for part in ('bin', 'runtime', 'config', 'home', 'state'):
|
|
(self.root / part).mkdir(mode=0o700)
|
|
self.env = dict(os.environ, HOME=str(self.root / 'home'),
|
|
XDG_CONFIG_HOME=str(self.root / 'config'),
|
|
XDG_RUNTIME_DIR=str(self.root / 'runtime'),
|
|
TEST_STATE=str(self.root / 'state'),
|
|
HYPRLAND_INSTANCE_SIGNATURE='test-instance',
|
|
PATH=str(self.root / 'bin') + ':' + os.environ['PATH'])
|
|
for name in ('loginctl', 'hyprctl', 'sudo', 'sleep', 'udevadm', 'fake-fix', 'systemctl'):
|
|
p = self.root / 'bin' / name
|
|
p.write_text(MOCK)
|
|
p.chmod(0o755)
|
|
self.config = self.root / 'config/revive-monitor/config.env'
|
|
self.config.parent.mkdir()
|
|
self.config.write_text('HDMI_SETTLE_SECONDS=0\nVT_SWITCH_DELAY_SECONDS=0\nVT_RETURN_SETTLE_SECONDS=0\n')
|
|
|
|
def run_script(self, name='revive-monitor-fix.sh', args=('1',), **env):
|
|
return subprocess.run(['bash', str(ROOT / 'scripts' / name), *args],
|
|
env=dict(self.env, **env), text=True, capture_output=True, timeout=15)
|
|
|
|
def calls(self, name):
|
|
p = self.root / 'state/calls'
|
|
return [a[1:] for a in map(json.loads, p.read_text().splitlines()) if a[0] == name] if p.exists() else []
|
|
|
|
def switches(self):
|
|
return [a[-1] for a in self.calls('sudo') if '-l' not in a]
|
|
|
|
def test_success(self):
|
|
r = self.run_script()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertEqual(self.switches(), ['2', '1'])
|
|
self.assertIn('Recovery completed', r.stderr)
|
|
|
|
def test_invalid_vt(self):
|
|
for vt in ('', '0', '64', '01', 'word', '99999999999999999'):
|
|
with self.subTest(vt=vt):
|
|
self.assertNotEqual(self.run_script(args=(vt,)).returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_session_guards(self):
|
|
for env in ({'NO_SESSION': '1'}, {'SESSION_UID': '99999'}, {'SESSION_TYPE': 'x11'},
|
|
{'SESSION_SEAT': 'seat1'}, {'SESSION_DESKTOP': 'GNOME'}):
|
|
with self.subTest(env=env):
|
|
self.assertEqual(self.run_script(**env).returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_wrong_vt(self):
|
|
self.assertNotEqual(self.run_script(args=('2',)).returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_session_changes_during_settle(self):
|
|
self.assertEqual(self.run_script(CHANGE_SESSION='1').returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_ipc_and_output_guards(self):
|
|
for env in ({'HYPRLAND_INSTANCE_SIGNATURE': ''}, {'HYPR_FAIL': '1'},
|
|
{'OUTPUT': 'DP-1'}, {'BAD_JSON': '1'}):
|
|
with self.subTest(env=env):
|
|
self.assertNotEqual(self.run_script(**env).returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_permission_checked_before_leaving(self):
|
|
self.assertNotEqual(self.run_script(DENY_RETURN='1').returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_return_failure_retried_and_reported(self):
|
|
r = self.run_script(RETURN_FAIL_ONCE='1')
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(self.switches(), ['2', '1', '1'])
|
|
self.assertNotIn('Recovery completed', r.stderr)
|
|
|
|
def test_termination_returns_to_desktop(self):
|
|
r = self.run_script(SIGNAL_AFTER_SWITCH='1')
|
|
self.assertEqual(r.returncode, 143, r.stderr)
|
|
self.assertEqual(self.switches(), ['2', '1'])
|
|
|
|
def test_process_group_termination_returns_to_desktop(self):
|
|
proc = subprocess.Popen(['bash', str(ROOT / 'scripts/revive-monitor-fix.sh'), '1'],
|
|
env=dict(self.env, BLOCK_AFTER_SWITCH='1'),
|
|
text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
start_new_session=True)
|
|
try:
|
|
deadline = time.monotonic() + 5
|
|
while not (self.root / 'state/sleeping').exists() and time.monotonic() < deadline:
|
|
time.sleep(0.02)
|
|
self.assertTrue((self.root / 'state/sleeping').exists())
|
|
os.killpg(proc.pid, signal.SIGTERM)
|
|
_, stderr = proc.communicate(timeout=10)
|
|
self.assertEqual(proc.returncode, 143, stderr)
|
|
self.assertEqual(self.switches(), ['2', '1'])
|
|
finally:
|
|
if proc.poll() is None:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.communicate()
|
|
|
|
def test_session_changes_during_permission_check(self):
|
|
self.assertEqual(self.run_script(CHANGE_DURING_PERMISSION_CHECK='1').returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_lock_open_failure_aborts(self):
|
|
state = self.root / 'runtime/revive-monitor'
|
|
state.mkdir(mode=0o700)
|
|
(state / 'fix.lock').mkdir()
|
|
self.assertNotEqual(self.run_script().returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_sudo_failure_not_success(self):
|
|
r = self.run_script(SUDO_FAIL='1')
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn('Emergency return failed', r.stderr)
|
|
self.assertNotIn('Recovery completed', r.stderr)
|
|
|
|
def test_no_shared_tmp_fallback(self):
|
|
self.assertNotEqual(self.run_script(XDG_RUNTIME_DIR='').returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_unsafe_runtime_rejected(self):
|
|
(self.root / 'runtime').chmod(0o755)
|
|
self.assertNotEqual(self.run_script().returncode, 0)
|
|
|
|
def test_symlink_state_rejected(self):
|
|
(self.root / 'runtime/revive-monitor').symlink_to(self.root / 'state')
|
|
self.assertNotEqual(self.run_script().returncode, 0)
|
|
self.assertFalse((self.root / 'state/fix.lock').exists())
|
|
|
|
def test_invalid_config(self):
|
|
for entry in ('DEBOUNCE_SECONDS=-1', 'HDMI_SETTLE_SECONDS=9999',
|
|
'VT_SWITCH_DELAY_SECONDS=abc', 'MONITOR="HDMI-A-2,garbage"'):
|
|
self.config.write_text(entry + '\n')
|
|
with self.subTest(entry=entry):
|
|
self.assertNotEqual(self.run_script().returncode, 0)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def test_leading_zero_timing_is_decimal(self):
|
|
self.config.write_text('HDMI_SETTLE_SECONDS=08\n')
|
|
self.assertEqual(self.run_script().returncode, 0)
|
|
self.assertEqual(self.calls('sleep')[0], ['8'])
|
|
|
|
def test_lock_blocks_overlapping_recovery(self):
|
|
state = self.root / 'runtime/revive-monitor'
|
|
state.mkdir(mode=0o700)
|
|
with (state / 'fix.lock').open('w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
r = self.run_script()
|
|
self.assertEqual(r.returncode, 0)
|
|
self.assertIn('already running', r.stderr)
|
|
self.assertEqual(self.switches(), [])
|
|
|
|
def watcher(self, **env):
|
|
return self.run_script('revive-monitor-drm-watch.sh', (),
|
|
REVIVE_MONITOR_FIX_SCRIPT=str(self.root / 'bin/fake-fix'), **env)
|
|
|
|
def test_watcher_eof_requests_restart(self):
|
|
for status in ('0', '7'):
|
|
with self.subTest(status=status):
|
|
r = self.watcher(UDEV_EXIT=status, EVENT_COUNT='0')
|
|
self.assertEqual(r.returncode, 1)
|
|
self.assertIn(f'event stream ended (exit {status})', r.stderr)
|
|
|
|
def test_watcher_debounces_burst(self):
|
|
r = self.watcher(EVENT_COUNT='4')
|
|
self.assertEqual(r.returncode, 1) # Mock stream ends.
|
|
self.assertEqual(self.calls('fake-fix'), [['1']])
|
|
|
|
def test_watcher_reports_recovery_failure(self):
|
|
r = self.watcher(FIX_EXIT='17')
|
|
self.assertIn('failed (exit 17)', r.stderr)
|
|
self.assertNotIn('helper finished', r.stderr)
|
|
|
|
def test_watcher_ignores_prelogin_events(self):
|
|
self.watcher(NO_SESSION='1')
|
|
self.assertEqual(self.calls('fake-fix'), [])
|
|
|
|
def test_watcher_missing_helper_aborts(self):
|
|
r = self.run_script('revive-monitor-drm-watch.sh', (),
|
|
REVIVE_MONITOR_FIX_SCRIPT=str(self.root / 'missing'))
|
|
self.assertEqual(r.returncode, 1)
|
|
self.assertEqual(self.calls('udevadm'), [])
|
|
|
|
@unittest.skipIf(os.geteuid() == 0, 'Installer intentionally rejects root')
|
|
def test_install_update_uninstall(self):
|
|
before = self.config.read_bytes()
|
|
for _ in range(2):
|
|
r = self.run_script('install.sh', ())
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertEqual(self.config.read_bytes(), before)
|
|
self.assertTrue((self.root / 'config/systemd/user/revive-monitor-drm-watch.service').exists())
|
|
helper = self.root / 'home/.local/bin/revive-monitor-fix'
|
|
self.assertTrue(os.access(helper, os.X_OK))
|
|
r = subprocess.run([str(helper), '1'], env=self.env, text=True, capture_output=True, timeout=15)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertNotIn('enable', str(self.calls('systemctl')))
|
|
r = self.run_script('install.sh', ('--start',), SERVICE_ACTIVE='1')
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertIn(['--user', 'restart', 'revive-monitor-drm-watch.service'], self.calls('systemctl'))
|
|
r = self.run_script('uninstall.sh', ())
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertFalse(helper.exists())
|
|
self.assertEqual(self.config.read_bytes(), before)
|
|
self.assertNotIn(['--user', 'reset-failed'], self.calls('systemctl'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|