"""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')}])) elif args == ['-j', 'status']: if os.environ.get('OLD_HYPRLAND'): print('unknown request') sys.exit(int(os.environ.get('STATUS_EXIT', '0'))) print(json.dumps({'configProvider': os.environ.get('CONFIG_PROVIDER', 'hyprlang')})) elif args[0] == 'dispatch': monitor = os.environ.get('OUTPUT', 'HDMI-A-2') if os.environ.get('CONFIG_PROVIDER') == 'lua': commands = { ('dispatch', f'hl.dsp.dpms({{ action = "on", monitor = "{monitor}" }})'): 'DPMS', ('dispatch', f'hl.dsp.focus({{ monitor = "{monitor}" }})'): 'focus', } else: commands = { ('dispatch', 'dpms', 'on', monitor): 'DPMS', ('dispatch', 'focusmonitor', monitor): 'focus', } action = commands.get(tuple(args)) if action is None: print('error: invalid dispatch syntax') sys.exit(7) if os.environ.get('DISPATCH_FAIL') == action and ( not os.environ.get('FINAL_REFRESH_FAIL') or (state / 'away').exists()): print(f'error: {action} rejected for {monitor}') sys.exit(int(os.environ.get('DISPATCH_EXIT', '7'))) print('ok') else: print('unexpected hyprctl call', file=sys.stderr) sys.exit(99) 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_lua_session_refreshes_before_and_after_vt_switch(self): r = self.run_script(CONFIG_PROVIDER='lua') self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(self.switches(), ['2', '1']) self.assertIn('Using lua dispatch syntax', r.stderr) dispatches = [a for a in self.calls('hyprctl') if a[0] == 'dispatch'] self.assertEqual(dispatches, [ ['dispatch', 'hl.dsp.dpms({ action = "on", monitor = "HDMI-A-2" })'], ['dispatch', 'hl.dsp.focus({ monitor = "HDMI-A-2" })'], ] * 2) def test_legacy_without_status_command(self): for code in ('0', '1'): with self.subTest(code=code): r = self.run_script(OLD_HYPRLAND='1', STATUS_EXIT=code) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('Using hyprlang dispatch syntax', r.stderr) def test_unknown_config_provider_aborts(self): r = self.run_script(CONFIG_PROVIDER='unknown') self.assertNotEqual(r.returncode, 0) self.assertIn('Unsupported Hyprland config provider', r.stderr) self.assertEqual(self.switches(), []) def test_dispatch_errors_reported_before_switch(self): for provider in ('lua', 'hyprlang'): for action in ('DPMS', 'focus'): for code in ('0', '7', '124'): with self.subTest(provider=provider, action=action, code=code): r = self.run_script(CONFIG_PROVIDER=provider, DISPATCH_FAIL=action, DISPATCH_EXIT=code) self.assertNotEqual(r.returncode, 0) self.assertIn(f'Hyprland {action} failed (exit {code})', r.stderr) self.assertIn(f'error: {action} rejected for HDMI-A-2', r.stderr) self.assertNotIn('Recovery completed', r.stderr) self.assertEqual(self.switches(), []) def test_final_lua_refresh_failure_not_reported_as_success(self): r = self.run_script(CONFIG_PROVIDER='lua', DISPATCH_FAIL='focus', FINAL_REFRESH_FAIL='1') self.assertNotEqual(r.returncode, 0) self.assertEqual(self.switches(), ['2', '1']) self.assertIn('Final Hyprland refresh failed', r.stderr) self.assertIn('error: focus rejected for HDMI-A-2', r.stderr) self.assertNotIn('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()