#!/usr/bin/env python3
"""Exercise the display API in a private compositor, never the active desktop."""
import json, os, socket, struct, subprocess, tempfile, time
from pathlib import Path

stage=Path(os.environ.get('WCM_TEST_STAGE','/var/tmp/wayfire-all-floating'))
runtime=Path(tempfile.mkdtemp(prefix='window-scope-test-'))
env=os.environ.copy()
for key in ['WAYLAND_DISPLAY','DISPLAY','WAYFIRE_SOCKET','_WAYFIRE_SOCKET','WAYFIRE_CONFIG_FILE','WAYFIRE_DEFAULT_CONFIG_BACKEND']:
    env.pop(key,None)
env.update(GIO_USE_VFS='local',GTK_USE_PORTAL='0',XDG_RUNTIME_DIR=str(runtime),WLR_BACKENDS='headless',WLR_HEADLESS_OUTPUTS='4',WLR_RENDERER='gles2',
    WLR_RENDER_DRM_DEVICE='/dev/dri/renderD128',WAYFIRE_PLUGIN_PATH=str(runtime)+':'+str(stage/'plugins'),WAYFIRE_PLUGIN_XML_PATH=str(stage/'metadata'))
import shutil
shutil.copy2(stage/'old-wm-actions.so', runtime/'libwm-actions.so')
config=runtime/'wayfire.ini'
config.write_text('''# Preserve this comment and unrelated settings.
[core]
plugins = ipc stipc ipc-rules output-control grid place move resize wm-actions simple-tile cube vswitch
xwayland = false
stitch_enabled = true
stitch_outputs = HEADLESS-1 HEADLESS-2 HEADLESS-3
default_output = VIRTUAL-1
window_scope = physical
window_scope_rules = [{"app_id":"scope-wide*","scope":"virtual"},{"app_id":"scope-pinned","scope":"HEADLESS-3"}]
[simple-tile]
key_toggle = <super> KEY_ENTER
tile_by_default = none
float_on_top = false
[wm-actions]
keep_floating_above = false
toggle_always_on_top = <super> KEY_T
[cube]
rotate_left = <alt> <ctrl> KEY_LEFT
rotate_right = <alt> <ctrl> KEY_RIGHT
[autostart]
autostart_wf_shell = false
[grid]
type = none
zone_gap = 15
[output:HEADLESS-1]
mode = 640x720@60
position = 0,0
[output:HEADLESS-2]
mode = 640x720@60
position = 640,0
[output:HEADLESS-3]
mode = 640x720@60
position = 1280,0
[output:HEADLESS-4]
mode = 320x240@60
position = 1920,0
masked = true
''')
log=(runtime/'wayfire.log').open('w')
compositor_env=env.copy()
if os.environ.get('WCM_TEST_ASAN'):
    compositor_env.update(LD_PRELOAD=os.environ['WCM_TEST_ASAN'],ASAN_OPTIONS='detect_leaks=0:abort_on_error=1',UBSAN_OPTIONS='halt_on_error=1:print_stacktrace=1')
process=subprocess.Popen([os.environ.get('WCM_TEST_COMPOSITOR','/usr/bin/wayfire'),'-c',str(config)],env=compositor_env,stdout=log,stderr=log)
def wait(fn,timeout=10):
    deadline=time.monotonic()+timeout
    while time.monotonic()<deadline:
        if process.poll() is not None:raise RuntimeError(f'Compositor exited; {runtime}/wayfire.log')
        result=fn()
        if result:return result
        time.sleep(.05)
    raise TimeoutError(f'Condition timed out; {runtime}/wayfire.log')
def ipc(method,data=None,fail=False):
    with socket.socket(socket.AF_UNIX) as s:
        s.settimeout(8);s.connect(env['WAYFIRE_SOCKET'])
        b=json.dumps({'method':method,'data':data or {}}).encode();s.sendall(struct.pack('<I',len(b))+b)
        def read(n):
            b=b''
            while len(b)<n:
                c=s.recv(n-len(b))
                if not c:raise RuntimeError('IPC disconnected')
                b+=c
            return b
        result=json.loads(read(struct.unpack('<I',read(4))[0]))
    if fail:assert result.get('error'),result
    else:assert not isinstance(result,dict) or not result.get('error'),result
    return result
def get():return ipc('displays/get')
def apply(options,fail=False):return ipc('displays/apply',{'revision':get()['revision'],'options':options},fail)
def revert(t):return ipc('displays/revert',{'token':t['token']})
def confirm(t,fail=False):return ipc('displays/confirm',{'token':t['token']},fail)
def names():return [d['name'] for d in get()['desktops']]

clients=[]
def spawn(app,mode='normal'):
    f=(runtime/(app+'.log')).open('w')
    p=subprocess.Popen([str(stage/'scope-client'),app,mode],env=env,stdin=subprocess.PIPE,stdout=f,stderr=f,text=True)
    clients.append((p,f))
    if mode=='bar':
        wait(lambda:any(v.get('app-id')=='scope-bar' and v.get('mapped') for v in ipc('window-rules/list-views')))
        return p,None
    v=wait(lambda:next((v for v in ipc('window-rules/list-views') if v.get('app-id')==app and v.get('mapped')),None))
    return p,v['id']
def info(vid):return next(v for v in ipc('window-rules/list-views') if v['id']==vid)
def geometry(vid):return info(vid)['geometry']
def command(p,cmd):p.stdin.write(cmd+'\n');p.stdin.flush()
def expect(vid,**wanted):
    try:return wait(lambda:all(geometry(vid)[k]==v for k,v in wanted.items()))
    except Exception:
        print('FAILED expected',wanted,'actual',info(vid),flush=True);raise

def focus(vid):ipc('window-rules/focus-view',{'id':vid})
def keys(*keys):
    for k in keys:ipc('stipc/feed_key',{'key':k,'state':True})
    for k in reversed(keys):ipc('stipc/feed_key',{'key':k,'state':False})
def layout():return ipc('simple-tile/get-layout',{'wset-index':wset,'workspace':{'x':0,'y':0}})['layout']
def contains(node,vid):
    return node.get('view-id')==vid or any(contains(c,vid) for c in node.get('vertical-split',[])+node.get('horizontal-split',[]))
def tiled(vid):return contains(layout(),vid)
def toggle(vid):
    was=tiled(vid);focus(vid);keys('KEY_LEFTMETA','KEY_ENTER');wait(lambda:tiled(vid)!=was)
def overlap():
    for vid in [aid,bid]:
        ipc('window-rules/configure-view',{'id':vid,'geometry':{'x':60,'y':80,'width':300,'height':180}})
        expect(vid,x=60,y=80,width=300,height=180)
def pixel(expected,name):
    from PIL import Image
    time.sleep(.15)
    path=runtime/(name+'.png')
    subprocess.run(['grim','-o','VIRTUAL-1',str(path)],env=env,check=True)
    actual=Image.open(path).getpixel((150,150))[:3]
    assert actual==expected,(name,actual,expected)
def options(opts):ipc('wayfire/set-config-options',opts)
try:
    env['WAYFIRE_SOCKET']=str(wait(lambda:next(runtime.glob('wayfire-*.socket'),None)))
    env['WAYLAND_DISPLAY']=next(p.name for p in runtime.glob('wayland-*') if not p.name.endswith('.lock'))
    print('Runtime:',runtime,flush=True)
    wait(lambda:names()==['VIRTUAL-1'])
    a,aid=spawn('float-target');b,bid=spawn('float-cover','maximize')
    wset=info(aid)['wset-index']
    assert not info(aid)['always-on-top'] and not info(bid)['always-on-top']
    (runtime/'libwm-actions.so').unlink()
    ipc('wayfire/reload-plugins')
    options({'wm-actions/keep_floating_above':True})
    wait(lambda:info(aid)['always-on-top'])
    assert not info(bid)['always-on-top']
    print('PASS live upgrade from resident old plugin; existing floats raised, maximized window stays below',flush=True)
    ipc('window-rules/configure-view',{'id':aid,'geometry':{'x':60,'y':80,'width':300,'height':180}})
    focus(bid);pixel((34,68,170),'floating-above-maximized-cover')
    c,cid=spawn('float-new')
    wait(lambda:info(cid)['always-on-top'])
    c.terminate();c.wait(timeout=5)
    print('PASS new floats on top; focusing a maximized window cannot cover a float',flush=True)
    toggle(aid);wait(lambda:not info(aid)['always-on-top'])
    assert tiled(aid)
    b.stdin.write('maximize 0\n');b.stdin.flush()
    wait(lambda:info(bid)['always-on-top'])
    ipc('window-rules/configure-view',{'id':bid,'geometry':{'x':60,'y':80,'width':300,'height':180}})
    focus(aid);pixel((221,85,51),'retiled-below-floating-cover')
    b.stdin.write('maximize 1\n');b.stdin.flush()
    wait(lambda:not info(bid)['always-on-top'])
    toggle(aid);wait(lambda:info(aid)['always-on-top'])
    print('PASS Super+Enter retains tiling membership, drops automatic above and floats back on top',flush=True)
    a.stdin.write('maximize 1\n');a.stdin.flush()
    wait(lambda:info(aid)['tiled-edges']==15 and not info(aid)['always-on-top'])
    a.stdin.write('maximize 0\n');a.stdin.flush()
    wait(lambda:info(aid)['tiled-edges']==0 and info(aid)['always-on-top'])
    ipc('wm-actions/set-fullscreen',{'id':aid,'state':True})
    wait(lambda:info(aid)['fullscreen'] and not info(aid)['always-on-top'])
    ipc('wm-actions/set-fullscreen',{'id':aid,'state':False})
    wait(lambda:not info(aid)['fullscreen'] and info(aid)['always-on-top'])
    print('PASS maximize/fullscreen clear automatic above; returning to floating restores it',flush=True)
    ipc('wm-actions/set-minimized',{'id':aid,'state':True})
    wait(lambda:info(aid)['minimized'])
    ipc('wm-actions/set-minimized',{'id':aid,'state':False})
    wait(lambda:not info(aid)['minimized'] and info(aid)['always-on-top'])
    focus(aid);keys('KEY_LEFTMETA','KEY_T')
    wait(lambda:not info(aid)['always-on-top'])
    ipc('wm-actions/set-minimized',{'id':aid,'state':True})
    ipc('wm-actions/set-minimized',{'id':aid,'state':False})
    time.sleep(.2);assert not info(aid)['always-on-top']
    toggle(aid);wait(lambda:info(aid)['tiled-edges']==15)
    toggle(aid);wait(lambda:info(aid)['always-on-top'])
    print('PASS minimize/restore preserves policy and manual Super+T override until retile',flush=True)
    ipc('simple-tile/set-layout',{'wset-index':wset,'workspace':{'x':0,'y':0},'layout':{'vertical-split':[{'view-id':aid,'weight':1.0}]}})
    wait(lambda:tiled(aid) and not info(aid)['always-on-top'])
    toggle(aid);wait(lambda:info(aid)['always-on-top'])
    options({'wm-actions/keep_floating_above':False})
    wait(lambda:not info(aid)['always-on-top'])
    ipc('wm-actions/set-always-on-top',{'id':bid,'state':True})
    options({'wm-actions/keep_floating_above':True})
    wait(lambda:info(aid)['always-on-top'])
    assert info(bid)['always-on-top']
    options({'wm-actions/keep_floating_above':False})
    wait(lambda:not info(aid)['always-on-top'])
    assert info(bid)['always-on-top']
    print('PASS IPC retiling and option reload; explicit on-top choice is preserved',flush=True)
finally:
    for p,f in clients:
        if p.poll() is None:p.terminate()
        p.wait(timeout=5);f.close()
    process.terminate();process.wait(timeout=10);log.close()