#!/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-physical-grid'))
runtime=Path(tempfile.mkdtemp(prefix='scope-toggle-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(stage/'plugins'),WAYFIRE_PLUGIN_XML_PATH=str(stage/'metadata'))
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":"tile-wide*","scope":"virtual"},{"app_id":"scope-pinned","scope":"HEADLESS-3"}]
[simple-tile]
key_toggle = none
tile_by_default = none
float_on_top = false
inner_gap_size = 10
outer_horiz_gap_size = 5
outer_vert_gap_size = 5
[wm-actions]
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]
toggle_scope = <super> KEY_ENTER
type = none
zone_gap = 20
[output:HEADLESS-1]
mode = 640x720@60
position = 0,0
[output:HEADLESS-2]
zone_gap = 20
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
''')
if os.environ.get('GRID_TEST_UPGRADE'):
    config.write_text(config.read_text().replace('output-control grid place', 'output-control '+str(stage/'libgrid-old.so')+' '+str(stage/'libgrid-duplicate.so')+' place'))
if os.environ.get('SCOPE_TEST_ANIMATION'):
    config.write_text(config.read_text().replace('type = none', 'type = crossfade\nduration = 300ms circle'))
if os.environ.get('GRID_TEST_UPGRADE'):
    config.write_text(config.read_text().replace(' place move ', ' place '+str(stage/'libmove-old.so')+' '))
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(*ks):
    for k in ks:ipc('stipc/feed_key',{'key':k,'state':True})
    for k in reversed(ks):ipc('stipc/feed_key',{'key':k,'state':False})
def layout(monitor,ws=0):return ipc('simple-tile/get-layout',{'wset-index':wset,'workspace':{'x':ws,'y':0},'monitor':monitor})['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 any(contains(layout(m),vid) for m in ['HEADLESS-1','HEADLESS-2','HEADLESS-3','VIRTUAL-1'])
def toggle(vid):
    was=tiled(vid);focus(vid);keys('KEY_LEFTMETA','KEY_ENTER');wait(lambda:tiled(vid)!=was)
def spawn_at(app,monitor):
    ipc('stipc/move_cursor',{'x':(monitor-1)*640+200,'y':200})
    return spawn(app)
def bounds(vid,x,y,w,h):
    g=geometry(vid);assert g['x']>=x and g['y']>=y and g['x']+g['width']<=x+w and g['y']+g['height']<=y+h,(vid,g,(x,y,w,h))
def toggle_scope(vid, keyboard=False):
    was=info(vid)['window-scope']
    if keyboard:
        focus(vid);keys('KEY_LEFTMETA','KEY_ENTER')
    else:ipc('grid/toggle_scope',{'view-id':vid})
    wait(lambda:(info(vid)['window-scope']=='virtual') != (was=='virtual'))
    time.sleep(.4 if os.environ.get('SCOPE_TEST_ANIMATION') else .1)

def same_geometry(vid, before):
    expect(vid,**before)

def rect(x,y,w,h):return dict(x=x,y=y,width=w,height=h)
def slot_rect(area,slot):
    x,y,w,h=(area[k] for k in ('x','y','width','height'))
    if slot%3==1:w=int(w/2)
    elif slot%3==0:x+=int(w/2);w=int(w/2)
    if slot>=7:h=int(h/2)
    elif slot<=3:y+=int(h/2);h=int(h/2)
    return rect(x,y,w,h)
def move(x,y):ipc('stipc/move_cursor',{'x':x,'y':y});time.sleep(.04)
def button(mode):ipc('stipc/feed_button',{'combo':'S-BTN_LEFT','mode':mode})
def prepare(vid,monitor=1):
    ipc('window-rules/set-window-scope',{'id':vid,'scope':'physical'})
    ipc('grid/restore',{'view-id':vid})
    g=rect((monitor-1)*640+140,180,300,180)
    ipc('window-rules/configure-view',{'id':vid,'geometry':g})
    expect(vid,**g);time.sleep(.4 if os.environ.get("SCOPE_TEST_ANIMATION") else .05);focus(vid)
def begin(vid,offset=150):
    g=geometry(vid);move(g['x']+offset,g['y']+60);button('press')
    move(g['x']+offset+30,g['y']+60)
def preview(expected):
    def found():
        return any(v.get('mapped') and v.get('type')!='toplevel' and v.get('app-id')!='scope-bar' and v.get('geometry')==expected for v in ipc('window-rules/list-views'))
    try:wait(found,3)
    except Exception:
        print('PREVIEW expected',expected,'views',ipc('window-rules/list-views'),flush=True);raise

def end(vid,expected):
    button('release');expect(vid,**expected);time.sleep(.25)
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'])
    bar,_=spawn('test-bar','bar')
    a,aid=spawn_at('grid-normal',1)
    if os.environ.get('GRID_TEST_UPGRADE'):
        ipc('grid/slot_r',{'view-id':aid});expect(aid,**rect(320,44,300,656))
        before=geometry(aid)
        config.write_text(config.read_text().replace(str(stage/'libgrid-old.so')+' '+str(stage/'libgrid-duplicate.so'),'grid').replace(str(stage/'libmove-old.so'),'move'))
        wait(lambda:all(str(stage/'plugins'/lib) in Path('/proc/'+str(process.pid)+'/maps').read_text() for lib in ['libgrid.so','libmove.so']))
        time.sleep(.4);expect(aid,**before)
        print('PASS replacing duplicate Grid and old Move preserves existing window geometry',flush=True)
    # Every edge/corner on every horizontal monitor, including both sides of seams.
    for monitor in range(1,4):
        x=(monitor-1)*640;area=rect(x+20,44,600,656)
        targets=[(x+1,300,4),(x+638,300,6),(x+320,1,5),(x+320,718,2),
                 (x+1,1,7),(x+638,1,9),(x+1,718,1),(x+638,718,3)]
        for px,py,slot in targets:
            prepare(aid,monitor);begin(aid);move(px,py)
            expected=slot_rect(area,slot);preview(expected);end(aid,expected)
            assert info(aid)['physical-monitor']==f'HEADLESS-{monitor}',info(aid)
        print(f'PASS all eight edge/corner previews and drops on monitor {monitor}',flush=True)
    # Same slot on a different monitor must update its preview before release.
    prepare(aid);begin(aid);move(320,1);preview(rect(20,44,600,656))
    move(960,1);preview(rect(660,44,600,656));end(aid,rect(660,44,600,656))
    print('PASS preview moves between monitors while staying in the same slot',flush=True)
    # Off-center grabs: cursor and window center choose different monitors.
    for start,offset,destx,slot,monitor in [(1,290,642,4,2),(2,10,638,6,1)]:
        prepare(aid,start);begin(aid,offset);move(destx,300)
        expected=slot_rect(rect((monitor-1)*640+20,44,600,656),slot)
        preview(expected);end(aid,expected)
        assert info(aid)['physical-monitor']==f'HEADLESS-{monitor}'
        ipc('grid/slot_c',{'view-id':aid});expect(aid,**rect((monitor-1)*640+20,44,600,656))
    print('PASS seam drops use the pointer monitor and subsequent keyboard Grid keeps it',flush=True)
    prepare(aid,2)
    ipc('grid/slot_r',{'view-id':aid});expect(aid,**rect(960,44,300,656))
    toggle_scope(aid,True);expect(aid,**rect(960,24,960,696))
    toggle_scope(aid,True);expect(aid,**rect(960,44,300,656))
    print('PASS keyboard Grid defaults physical; Super+Enter explicitly toggles VIRTUAL',flush=True)
    prepare(aid,2);ipc('grid/use_virtual',{'view-id':aid});begin(aid);move(960,1)
    preview(rect(0,24,1920,696));end(aid,rect(0,24,1920,696))
    prepare(aid,2);ipc('grid/use_virtual',{'view-id':aid});begin(aid);move(1918,718)
    preview(rect(960,372,960,348));end(aid,rect(960,372,960,348))
    print('PASS explicit VIRTUAL dragging previews and drops across the combined desktop',flush=True)
    # Shift zone-fill remains a deliberate full-monitor shortcut.
    prepare(aid);begin(aid);move(1500,300)
    ipc('stipc/feed_key',{'key':'KEY_LEFTSHIFT','state':True});preview(rect(1300,44,600,656))
    end(aid,rect(1300,44,600,656));ipc('stipc/feed_key',{'key':'KEY_LEFTSHIFT','state':False})
    ipc('grid/slot_l',{'view-id':aid});expect(aid,**rect(1300,44,300,656))
    print('PASS Shift zone-fill retains the destination physical monitor',flush=True)
    # Moving into the middle of a monitor cancels the preview and stays floating.
    prepare(aid);begin(aid);move(638,300);preview(rect(320,44,300,656))
    move(960,300);button('release');time.sleep(.3)
    assert info(aid)['tiled-edges']==0,info(aid)
    assert not any(v.get('mapped') and v.get('pid')==-1 for v in ipc('window-rules/list-views'))
    print('PASS leaving an edge cancels preview and keeps the drop floating',flush=True)
    # Fullscreen remains fullscreen across an edge drag.
    prepare(aid);command(a,'fullscreen 1');expect(aid,**rect(0,0,640,720));time.sleep(.4)
    begin(aid);move(638,718);button('release');time.sleep(.3)
    assert info(aid)['fullscreen'] and info(aid)['tiled-edges']==0,info(aid)
    command(a,'fullscreen 0');time.sleep(.2)
    print('PASS dragging fullscreen does not apply Grid tiling',flush=True)
    # Independent outputs work at vertical global offsets too.
    prepare(aid)
    t=apply({'core/stitch_enabled':'false','core/default_output':'','output:HEADLESS-2/position':'0,720',
             'output:HEADLESS-3/position':'0,1440'});confirm(t);time.sleep(.3)
    ipc('window-rules/set-window-scope',{'id':aid,'scope':'HEADLESS-2'})
    ipc('grid/restore',{'view-id':aid})
    ipc('window-rules/configure-view',{'id':aid,'geometry':rect(140,180,300,180)})
    expect(aid,**rect(140,180,300,180));focus(aid)
    move(290,960);button('press');move(638,1438)
    preview(rect(320,360,320,360));end(aid,rect(320,360,320,360))
    ipc('grid/slot_r',{'view-id':aid});expect(aid,**rect(320,0,320,720))
    print('PASS unstitched output at a vertical offset: preview, drop and keyboard Grid',flush=True)
    print('ALL PASS',flush=True)
finally:
    for p,f in clients:
        if p.poll() is None:p.terminate()
        try:p.wait(timeout=3)
        except subprocess.TimeoutExpired:p.kill();p.wait()
        f.close()
    process.terminate()
    try:process.wait(timeout=5)
    except subprocess.TimeoutExpired:process.kill();process.wait()
    log.close()