feresaria/scripts/tests/test_launcher_rime_isolated.py

"""Real shell + native Rime test on a private Xvfb/nested Wayland/bus.
No host GUI input or application profile is connected to this fixture.

FERESE_TEST_LAUNCHER_IME=1 python3 scripts/tests/test_launcher_rime_isolated.py
Build the release compositor/shell/Rime/feresectl and debug ferese-rime-probe first.
Synthetic screenshots and protocol traces stay under .artifacts/.
"""
from pathlib import Path
import hashlib,json,os,re,secrets,signal,shutil,subprocess,tempfile,time,unittest

REPO=Path(__file__).resolve().parents[2]
BASE=REPO/'.artifacts/launcher-rime-isolated'
COMPOSITOR=REPO/os.environ.get('FERESE_TEST_BINARY','target/release/ferese')
PROBE=REPO/os.environ.get('FERESE_TEST_PROBE','target/debug/ferese-rime-probe')
CTL=REPO/os.environ.get('FERESE_TEST_CTL','target/release/feresectl')
RIME=REPO/os.environ.get('FERESE_TEST_RIME_BINARY','target/release/ferese-rime')

def wait(fn,label,timeout=20):
 end=time.monotonic()+timeout
 while time.monotonic()<end:
  value=fn()
  if value:return value
  time.sleep(.05)
 raise RuntimeError('Timed out: '+label)

def run_case(label,shell,xdisplay,xauth):
 case=BASE/label;case.mkdir(mode=0o700,exist_ok=True)
 runtime=Path(tempfile.mkdtemp(prefix='flime-',dir='/run/user/1000'))
 for name in ['config/ferese','data','state','cache','home','rime','tmp']:(case/name).mkdir(mode=0o700,parents=True,exist_ok=True)
 (case/'config/ferese/config.kdl').write_text('xwayland { enabled #false; }\nanimations { enabled #false; }\nstatus { keybinding-guide #false; }\n')
 (case/'rime/default.custom.yaml').write_text('patch:\n  schema_list:\n    - schema: luna_pinyin_simp\n')
 conf=runtime/'bus.conf';conf.write_text('<busconfig><type>session</type><listen>unix:path='+str(runtime/'bus')+'</listen><policy context="default"><allow send_destination="*"/><allow receive_sender="*"/><allow own="*"/></policy></busconfig>')
 env={k:v for k,v in os.environ.items() if k not in ['WAYLAND_DISPLAY','WAYLAND_SOCKET','FERESE_SOCKET','FERESE_SHELL_CONTROL_SOCKET','DBUS_SESSION_BUS_ADDRESS','GTK_IM_MODULE','QT_IM_MODULE','IBUS_ADDRESS','XMODIFIERS']}
 env.update(HOME=str(case/'home'),XDG_RUNTIME_DIR=str(runtime),XDG_CONFIG_HOME=str(case/'config'),XDG_DATA_HOME=str(case/'data'),XDG_CACHE_HOME=str(case/'cache'),XDG_STATE_HOME=str(case/'state'),XDG_DATA_DIRS='/usr/local/share:/usr/share',DISPLAY=xdisplay,XAUTHORITY=str(xauth),TMPDIR=str(case/'tmp'),FERESE_ENABLE_INPUT_METHOD='1',FERESE_ENABLE_SCREENCOPY='1',FERESE_SESSION_IMPORT_ENV='0',FERESE_RIME_CONTROL=str(runtime/'rime.sock'),LIBGL_ALWAYS_SOFTWARE='1',__EGL_VENDOR_LIBRARY_FILENAMES='/usr/share/glvnd/egl_vendor.d/50_mesa.json',WINIT_UNIX_BACKEND='x11',ICED_BACKEND='tiny-skia',PATH='/usr/bin',RUST_LOG='ferese=info')
 children=[];log=(case/'session.log').open('w')
 try:
  bus=subprocess.Popen(['dbus-daemon','--config-file='+str(conf),'--nofork','--print-address=1'],env=env,stdout=subprocess.PIPE,stderr=log,text=True,start_new_session=True);children.append(bus)
  env['DBUS_SESSION_BUS_ADDRESS']=bus.stdout.readline().strip();assert env['DBUS_SESSION_BUS_ADDRESS'].startswith('unix:')
  comp=subprocess.Popen([str(COMPOSITOR),'--backend','nested','--supervise-client','--grant-effects','--grant-shell-control','--','/usr/bin/env','WAYLAND_DEBUG=client',str(shell)],env=env,stdout=log,stderr=log,start_new_session=True);children.append(comp)
  def find_socket():
   candidates=list(runtime.glob('ferese/instances/*/control.sock'))
   return next((p for p in candidates if p.is_socket()),None)
  sock=wait(find_socket,'private IPC')
  def ctl(*args):return subprocess.check_output([str(CTL),'--socket',str(sock),*map(str,args)],env=env,text=True)
  outputs=json.loads(ctl('-j','outputs'));assert len(outputs)==1 and outputs[0]['name']=='ferese-winit',outputs
  wayland=wait(lambda:next((p for p in runtime.glob('wayland-*') if p.is_socket()),None),'private Wayland')
  child_env=env|{'WAYLAND_DISPLAY':str(wayland),'FERESE_SOCKET':str(sock)}
  ime=subprocess.Popen([str(RIME),'--user-data',str(case/'rime'),'--shared-data','/usr/share/rime-data'],env=child_env,stdout=log,stderr=log,start_new_session=True);children.append(ime)
  wait(lambda:(runtime/'rime.sock').is_socket(),'native Rime')
  def status():return json.loads(subprocess.check_output([str(RIME),'--status'],env=child_env,text=True))['state']
  wait(lambda: 'ferese_shell_manager_v1' in (case/'session.log').read_text(), 'real shell control binding')
  time.sleep(.5);ctl('launcher','toggle')
  wait(lambda:status()['active'],'launcher text-input enable')
  subprocess.run([str(RIME),'--mode','rime'],env=child_env,capture_output=True,text=True,check=True)
  def cursor_x():
   return [int(x) for x in re.findall(r'zwp_text_input_v3[@#]\d+\.set_cursor_rectangle\((-?\d+),', (case/'session.log').read_text())]
  initial=wait(lambda:cursor_x(),'initial caret')[-1]
  testenv=child_env|{'FERESE_TEST_INPUT':'1'}
  def probe(*args):
   p=subprocess.run([str(PROBE),*args],env=testenv,capture_output=True,text=True,timeout=10)
   assert p.returncode==0,p.stderr
  probe('--keys','nihao');time.sleep(.3)
  (case/'preedit.png').write_bytes(subprocess.check_output([str(CTL),'--socket',str(sock),'screenshot'],env=env))
  probe('--keys',' ');time.sleep(.6)
  (case/'commit.png').write_bytes(subprocess.check_output([str(CTL),'--socket',str(sock),'screenshot'],env=env))
  trace=(case/'session.log').read_text();xs=cursor_x()
  result={'label':label,'shell':str(shell),'shellSha256':hashlib.sha256(shell.read_bytes()).hexdigest(),'compositorPid':comp.pid,'singlePrivateOutputVerified':True,'nativeRimePid':ime.pid,'launcherImeActive':status()['active'],'committedSyntheticHelloSeenOnProtocol':bool(re.search(r'commit_string\("你好"\)',trace)),'initialCaretX':initial,'caretXs':xs,'committedTextMovedCaret':bool(xs and xs[-1]>initial+8),'privateHeadlessX11Parent':xdisplay,'hostGuiUsed':False,'userProfilesUsed':False,'messagesSent':False}
  (case/'result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');print(json.dumps(result,ensure_ascii=False))
  probe('--chord','Control_L+a','--key','BackSpace','--key','Escape')
  return result
 finally:
  for child in reversed(children):
   if child.poll() is None:
    os.killpg(child.pid,signal.SIGTERM)
    try:child.wait(timeout=4)
    except subprocess.TimeoutExpired:os.killpg(child.pid,signal.SIGKILL);child.wait()
  log.close()
  # Keep synthetic schema/screenshots/logs as evidence; remove dead private sockets.
  import shutil
  shutil.rmtree(runtime)



@unittest.skipUnless(os.environ.get('FERESE_TEST_LAUNCHER_IME') == '1', 'opt-in isolated launcher/Rime test')
class LauncherRimeTest(unittest.TestCase):
 def test_preedit_and_commit_reach_launcher(self):
  for tool in ['Xvfb','xauth','dbus-daemon']:
   if shutil.which(tool) is None:self.skipTest('missing fixture tool: '+tool)
  shell=REPO/os.environ.get('FERESE_TEST_SHELL','target/release/ferese-shell')
  for binary in [COMPOSITOR,PROBE,CTL,RIME,shell]:
   if not binary.is_file():self.skipTest('build required binary: '+str(binary))
  BASE.mkdir(mode=0o700,parents=True,exist_ok=True)
  auth=BASE/'fixture.Xauthority';cookie=secrets.token_hex(16);xvfb=None
  with (BASE/'xvfb.log').open('w') as log:
   try:
    for number in range(190,230):
     if Path(f'/tmp/.X{number}-lock').exists() or Path(f'/tmp/.X11-unix/X{number}').exists():continue
     display=':'+str(number)
     subprocess.run(['xauth','-f',str(auth),'add',display,'MIT-MAGIC-COOKIE-1',cookie],capture_output=True,check=True)
     auth.chmod(0o600)
     xvfb=subprocess.Popen(['Xvfb',display,'-screen','0','1600x1000x24','-nolisten','tcp','-auth',str(auth)],stdout=log,stderr=log,start_new_session=True)
     try:wait(lambda:Path(f'/tmp/.X11-unix/X{number}').is_socket(),'Xvfb startup',5)
     except RuntimeError:
      if xvfb.poll() is None:os.killpg(xvfb.pid,signal.SIGTERM);xvfb.wait()
      xvfb=None;continue
     break
    self.assertIsNotNone(xvfb,'No private Xvfb display available')
    result=run_case('current',shell.resolve(),display,auth)
    self.assertTrue(result['singlePrivateOutputVerified'])
    self.assertTrue(result['committedSyntheticHelloSeenOnProtocol'])
    self.assertTrue(result['committedTextMovedCaret'], 'Wayland commit arrived but did not update launcher text/caret')
   finally:
    if xvfb is not None and xvfb.poll() is None:os.killpg(xvfb.pid,signal.SIGTERM);xvfb.wait(timeout=5)
    auth.unlink(missing_ok=True)

if __name__=='__main__':unittest.main()