108 lines
5.1 KiB
Python
108 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify selected colors in the actual VNC framebuffer, without VNC input."""
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import struct
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
loader = importlib.machinery.SourceFileLoader('adapter', '/usr/local/bin/lazyboy-cua-adapter-test')
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
adapter = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(adapter)
|
|
|
|
|
|
def capture():
|
|
with socket.create_connection(('127.0.0.1', 5900), 5) as stream:
|
|
def read(size):
|
|
data = bytearray()
|
|
while len(data) < size:
|
|
chunk = stream.recv(size - len(data))
|
|
assert chunk, 'VNC closed before a complete framebuffer'
|
|
data.extend(chunk)
|
|
return bytes(data)
|
|
assert read(12).startswith(b'RFB ')
|
|
stream.sendall(b'RFB 003.008\n')
|
|
assert 1 in read(read(1)[0]), 'disposable desktop must offer unauthenticated loopback VNC'
|
|
stream.sendall(b'\x01')
|
|
assert read(4) == b'\0' * 4
|
|
stream.sendall(b'\x01')
|
|
width, height = struct.unpack('>HH', read(4))
|
|
read(16)
|
|
read(struct.unpack('>I', read(4))[0])
|
|
assert 0 < width <= 4096 and 0 < height <= 4096
|
|
# Request raw pixels in RGBX byte order. Never send keys or pointer input.
|
|
stream.sendall(b'\0' * 4 + struct.pack('>BBBBHHHBBBxxx', 32, 24, 0, 1, 255, 255, 255, 0, 8, 16))
|
|
stream.sendall(struct.pack('>BBHi', 2, 0, 1, 0))
|
|
stream.sendall(struct.pack('>BBHHHH', 3, 0, 0, 0, width, height))
|
|
pixels = bytearray(width * height * 4)
|
|
while True:
|
|
kind = read(1)[0]
|
|
if kind == 2:
|
|
continue
|
|
if kind == 3:
|
|
read(3)
|
|
read(struct.unpack('>I', read(4))[0])
|
|
continue
|
|
assert kind == 0, kind
|
|
read(1)
|
|
for _ in range(struct.unpack('>H', read(2))[0]):
|
|
x, y, w, h, encoding = struct.unpack('>HHHHi', read(12))
|
|
assert encoding == 0 and x + w <= width and y + h <= height
|
|
rectangle = read(w * h * 4)
|
|
for row in range(h):
|
|
start = ((y + row) * width + x) * 4
|
|
pixels[start:start + w * 4] = rectangle[row * w * 4:(row + 1) * w * 4]
|
|
return width, height, pixels
|
|
|
|
|
|
def save_png(path, width, height, pixels):
|
|
pixels[3::4] = b'\xff' * (width * height)
|
|
rows = b''.join(b'\0' + pixels[y * width * 4:(y + 1) * width * 4] for y in range(height))
|
|
def chunk(kind, data):
|
|
return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data))
|
|
Path(path).write_bytes(b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 6, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(rows)) + chunk(b'IEND', b''))
|
|
|
|
|
|
adapter.wait_health()
|
|
root = Path('/tmp/lazyboy')
|
|
files = [root / 'screen-1.agent-name', root / 'screen-1.agent-color', root / 'screen-2.agent-color']
|
|
previous = {path: path.read_bytes() if path.exists() else None for path in files}
|
|
name = '小幫手 Alice'
|
|
try:
|
|
subprocess.run(['lazyboy-screen', 'ensure', '0', '', name, '#8B5CF6'], check=True)
|
|
adapter.api('/browser', {'action': 'snapshot', 'ensure': True})
|
|
adapter.api('/browser', {'action': 'navigate', 'url': 'about:blank', 'ensure': True})
|
|
pid = (root / 'screen-1-cua.pid').read_text()
|
|
for color, label in [('#8B5CF6', 'purple'), ('#22C55E', 'green'), ('#E11D48', 'pink')]:
|
|
subprocess.run(['lazyboy-screen', 'color', '0', color], check=True)
|
|
subprocess.run(['lazyboy-screen', 'color', '1', '#123456'], check=True)
|
|
assert (root / 'screen-1-cua.pid').read_text() == pid, 'color change restarted Cua'
|
|
assert files[0].read_text() == name, 'color change renamed the session'
|
|
adapter.api('/act', {'actions': [{'kind': 'pointer', 'type': 'click', 'x': 640, 'y': 400}], 'observe': False, 'settle_ms': 100})
|
|
expected = bytes.fromhex(color[1:])
|
|
deadline = time.monotonic() + 5
|
|
while True:
|
|
width, height, pixels = capture()
|
|
matches = sum(pixels[(y * width + x) * 4:(y * width + x) * 4 + 3] == expected
|
|
for y in range(380, 425) for x in range(620, 665))
|
|
if matches >= 3:
|
|
break
|
|
assert time.monotonic() < deadline, (color, 'selected RGB absent from VNC cursor')
|
|
time.sleep(0.1)
|
|
save_png(f'/tmp/cua-cursor-{label}.png', width, height, pixels)
|
|
print(f'VNC cursor {color}: {matches} exact RGB pixels, same session and daemon', flush=True)
|
|
rejected = subprocess.run(['lazyboy-screen', 'color', '0', 'red'], capture_output=True)
|
|
assert rejected.returncode != 0 and files[1].read_text() == '#E11D48'
|
|
finally:
|
|
adapter.smoke.cua_call('end_session', {'session': name})
|
|
for path, value in previous.items():
|
|
if value is None:
|
|
path.unlink(missing_ok=True)
|
|
else:
|
|
path.write_bytes(value)
|
|
print('Selected cursor colors, live updates, display isolation and invalid-color rejection passed')
|