100 lines
6.2 KiB
Python
100 lines
6.2 KiB
Python
"""Physical test against the separately started disposable logind session.
|
|
FERESE_TEST_DRM=1 python3 scripts/tests/test_rime_drm_isolated.py
|
|
Never starts/switches sessions itself. Requires its prepared lock-cycle binary.
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import time
|
|
import unittest
|
|
|
|
ROOT=Path(__file__).resolve().parents[2]
|
|
spec=importlib.util.spec_from_file_location("drm",ROOT/"scripts/rime-drm-validation.py")
|
|
d=importlib.util.module_from_spec(spec);spec.loader.exec_module(d)
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_DRM")=="1","explicit physical-session validation")
|
|
class PhysicalRime(unittest.TestCase):
|
|
def test_three_outputs_and_lock_cycle(self):
|
|
for _ in range(100):
|
|
if (d.DATA/"context.json").exists() and (d.DATA/"entry-state.json").exists():break
|
|
time.sleep(.1)
|
|
c=json.loads((d.DATA/"context.json").read_text())
|
|
self.assertTrue(Path(f"/proc/{c['compositor_pid']}").exists())
|
|
self.assertNotEqual(subprocess.check_output(["loginctl","show-seat","seat0","-p","ActiveSession","--value"],text=True).strip(),d.original())
|
|
env={**os.environ,"XDG_RUNTIME_DIR":str(d.RUNTIME),"WAYLAND_DISPLAY":c["wayland"],
|
|
"DBUS_SESSION_BUS_ADDRESS":c["bus"],"FERESE_SOCKET":c["socket"],"GDK_BACKEND":"wayland",
|
|
"GTK_IM_MODULE":"wayland","FERESE_TEST_INPUT":"1","FERESE_TEST_DRM_PID":str(c["compositor_pid"]),
|
|
"FERESE_RIME_CONTROL":str(d.RUNTIME/"rime-control.sock"),"GTK_A11Y":"none","GSETTINGS_BACKEND":"keyfile"}
|
|
env["GSK_RENDERER"]=os.environ.get("FERESE_TEST_GSK","vulkan")
|
|
control=[str(ROOT/"target/debug/feresectl"),"--socket",c["socket"]]
|
|
def run(args):return subprocess.check_output(args,env=env,text=True,stderr=subprocess.PIPE).strip()
|
|
def ctl(*args):return json.loads(run([str(ROOT/"target/debug/ferese-rime"),*args]))["state"]
|
|
def wait(fn,message,timeout=5):
|
|
until=time.monotonic()+timeout
|
|
while time.monotonic()<until:
|
|
if fn():return
|
|
time.sleep(.1)
|
|
raise AssertionError(message)
|
|
def keys(*args):run([str(ROOT/"target/debug/ferese-rime-probe"),*args]);time.sleep(.1)
|
|
outputs=json.loads(run(control+["--json","outputs"]))
|
|
active=[o for o in outputs if o["enabled"]]
|
|
self.assertEqual({o["name"] for o in active},{"DP-4","DP-5","DP-6"})
|
|
checks=[];apps=[]
|
|
(d.DATA/"outputs.json").write_text(json.dumps(outputs,indent=2)+"\n")
|
|
try:
|
|
for output in sorted(active,key=lambda o:o["x"]):
|
|
name=output["name"]; app=f"dev.ferese.RimeDRM{output['id']}";file=d.DATA/f"entry-{name}.json"
|
|
file.unlink(missing_ok=True)
|
|
run(control+["focus-output",name])
|
|
proc=subprocess.Popen(["gjs",str(ROOT/"scripts/tests/fixtures/rime-entry.js"),str(file),app],env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL);apps.append(proc)
|
|
wait(lambda:file.exists(),"Fresh GTK fixture did not initialize",timeout=10)
|
|
wait(lambda: (ctl("--status")["context"] or {}).get("app_id")==app,"Monitor fixture did not activate")
|
|
time.sleep(.25)
|
|
ctl("--mode","rime")
|
|
keys("--keys","nihao")
|
|
image=subprocess.check_output(control+["screenshot","--geometry",f"{output['x']},{output['y']} {output['width']}x{output['height']}"],env=env)
|
|
(d.DATA/f"candidate-{name}.png").write_bytes(image)
|
|
keys("--keys"," ")
|
|
wait(lambda:json.loads(file.read_text())["normal"]=="你好",f"Chinese input failed on {name}")
|
|
checks.append(f"{name}: Chinese preedit/candidates/commit at {output['scale']}x, {output['current_mode']['refresh_hz']} Hz")
|
|
|
|
keys("--keys","nihao")
|
|
release=d.DATA/"unlock-trigger";release.unlink(missing_ok=True)
|
|
log=(d.DATA/"lock-cycle.log").open("w")
|
|
locker=subprocess.Popen([str(d.DATA/"lock-cycle"),str(release)],env={**env,"FERESE_TEST_LOCK":"1"},stdout=log,stderr=subprocess.STDOUT)
|
|
try:
|
|
wait(lambda:'"event":"locked"' in (d.DATA/"lock-cycle.log").read_text(),"Physical outputs did not confirm protected presentation")
|
|
self.assertIn('"outputs":3',(d.DATA/"lock-cycle.log").read_text())
|
|
wait(lambda:not ctl("--status")["active"],"IME remained active while locked")
|
|
before=file.read_text();keys("--keys","secret-test")
|
|
self.assertEqual(file.read_text(),before)
|
|
screenshot=subprocess.run(control+["screenshot"],env=env,capture_output=True)
|
|
self.assertNotEqual(screenshot.returncode,0)
|
|
old=ctl("--status")["pid"]
|
|
subprocess.run(["systemctl","--user","kill","--signal=KILL",d.IME],env=d.manager_env(),check=True)
|
|
def recovered():
|
|
try:return ctl("--status")["pid"]!=old
|
|
except (subprocess.SubprocessError,ValueError):return False
|
|
wait(recovered,"IME did not recover during lock",timeout=8)
|
|
self.assertFalse(ctl("--status")["active"])
|
|
release.touch();self.assertEqual(locker.wait(timeout=5),0)
|
|
wait(lambda:ctl("--status")["active"],"Text input did not reactivate on unlock")
|
|
keys("--key","End","--keys","nihao ")
|
|
wait(lambda:json.loads(file.read_text())["normal"]=="你好你好","Chinese input failed after unlock")
|
|
checks.append("three physical outputs confirmed locked; candidates/input disabled; capture denied; IME restarted while locked; unlock restores Chinese")
|
|
finally:
|
|
release.touch()
|
|
if locker.poll() is None:locker.wait(timeout=5)
|
|
log.close()
|
|
(d.DATA/f"validation-{env['GSK_RENDERER']}.json").write_text(json.dumps({"status":"passed","renderer":env["GSK_RENDERER"],"checks":checks,"pam_authentication":"not covered by protocol-owner test"},ensure_ascii=False,indent=2)+"\n")
|
|
print(json.dumps(checks,ensure_ascii=False))
|
|
finally:
|
|
for proc in apps:
|
|
proc.terminate();proc.wait(timeout=5)
|
|
|
|
|
|
if __name__=="__main__":unittest.main()
|