mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-11 11:01:07 +00:00
105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
IME = Path(__file__).resolve().parents[1] / 'linuxfs/overlay/usr/local/bin/droiddeck-ime'
|
|
|
|
|
|
def message(obj, opcode, payload=b''):
|
|
return struct.pack('<II', obj, ((8 + len(payload)) << 16) | opcode) + payload
|
|
|
|
|
|
def wl_string(value):
|
|
data = value.encode() + b'\0'
|
|
return struct.pack('<I', len(data)) + data + b'\0' * (-len(data) % 4)
|
|
|
|
|
|
class DesktopImeTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.session = Path(self.temp.name)
|
|
self.server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.addCleanup(self.server.close)
|
|
env = dict(os.environ, BL_LAUNCH_DIR=str(self.session), WAYLAND_SOCKET=str(client.fileno()))
|
|
self.proc = subprocess.Popen(['python3', str(IME)], env=env, pass_fds=[client.fileno()],
|
|
stderr=subprocess.PIPE, text=True)
|
|
client.close()
|
|
self.addCleanup(self.stop)
|
|
self.server.settimeout(5)
|
|
self.assertEqual(self.read(12), message(1, 1, struct.pack('<I', 2)))
|
|
self.server.sendall(message(2, 0, struct.pack('<I', 7) + wl_string('zwp_input_method_v1') + struct.pack('<I', 1)))
|
|
bind = self.read(8 + 4 + len(wl_string('zwp_input_method_v1')) + 8)
|
|
self.assertEqual(bind, message(2, 0, struct.pack('<I', 7) + wl_string('zwp_input_method_v1') + struct.pack('<II', 1, 3)))
|
|
|
|
def stop(self):
|
|
if self.proc.poll() is None:
|
|
self.proc.kill()
|
|
self.proc.wait(5)
|
|
self.proc.stderr.close()
|
|
|
|
def read(self, size):
|
|
data = b''
|
|
while len(data) < size:
|
|
chunk = self.server.recv(size - len(data))
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
return data
|
|
|
|
def state(self, seq, timeout=5):
|
|
path = self.session / 'agent/ime.json'
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
snapshot = json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
snapshot = None
|
|
if snapshot and snapshot['seq'] >= seq:
|
|
return snapshot
|
|
time.sleep(0.02)
|
|
self.fail('ime.json never reached seq %d' % seq)
|
|
|
|
def test_focus_and_blur_are_reported(self):
|
|
self.server.sendall(message(3, 0, struct.pack('<I', 100)))
|
|
self.assertTrue(self.state(1)['textField'])
|
|
self.server.sendall(message(100, 2, struct.pack('<II', 0, 9)))
|
|
snapshot = self.state(2)
|
|
self.assertEqual((snapshot['textField'], snapshot['purpose']), (True, 9))
|
|
self.server.sendall(message(3, 1, struct.pack('<I', 100)))
|
|
self.assertEqual(self.read(8), message(100, 0))
|
|
self.assertFalse(self.state(3)['textField'])
|
|
self.assertEqual(oct((self.session / 'agent/ime.json').stat().st_mode & 0o777), '0o600')
|
|
|
|
def test_switching_fields_does_not_flicker(self):
|
|
self.server.sendall(message(3, 0, struct.pack('<I', 100)))
|
|
self.assertTrue(self.state(1)['textField'])
|
|
self.server.sendall(message(3, 1, struct.pack('<I', 100)) + message(3, 0, struct.pack('<I', 101)))
|
|
self.assertEqual(self.read(8), message(100, 0))
|
|
time.sleep(0.4)
|
|
self.assertEqual(self.state(1), {'textField': True, 'hint': 0, 'purpose': 0, 'seq': 1})
|
|
|
|
def test_compositor_exit_clears_the_field(self):
|
|
self.server.sendall(message(3, 0, struct.pack('<I', 100)))
|
|
self.assertTrue(self.state(1)['textField'])
|
|
self.server.close()
|
|
self.assertEqual(self.proc.wait(5), 0)
|
|
self.assertFalse(self.state(2)['textField'])
|
|
|
|
|
|
def test_a_protocol_error_clears_the_field(self):
|
|
self.server.sendall(message(3, 0, struct.pack('<I', 100)))
|
|
self.assertTrue(self.state(1)['textField'])
|
|
self.server.sendall(message(1, 0, struct.pack('<II', 3, 1) + wl_string('bad request')))
|
|
self.assertEqual(self.proc.wait(5), 1)
|
|
self.assertFalse(self.state(2)['textField'])
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|