"""Read-only review protocol checks in an isolated Wine prefix/session bus."""
import argparse
import os
from pathlib import Path
import subprocess
import sys
import time
from gi.repository import Gio, GLib

from wine_test_environment import add_test_options, run_wine_test, wine_test_directory

parser = argparse.ArgumentParser()
parser.add_argument('--build-dir', default='build')
parser.add_argument('--32-bit', dest='is32', action='store_true')
parser.add_argument('--session', action='store_true')
parser.add_argument('--reader', choices=('cthulhu', 'orca'))
parser.add_argument('--extension-endpoint', action='store_true')
add_test_options(parser)
args = parser.parse_args()
build = Path(args.build_dir).resolve()
if not args.session:
    sys.exit(run_wine_test(['dbus-run-session', '--', sys.executable, __file__,
                             '--session', '--build-dir', str(build),
                             *(['--extension-endpoint'] if args.extension_endpoint else []),
                             *(['--reader', args.reader] if args.reader else []),
                             *(['--32-bit'] if args.is32 else [])], build, "review_integration",
                              keep=args.keep, keepOnFailure=args.keepOnFailure))
busName = 'org.gnome.Orca1.Service' if args.reader == 'orca' else 'org.stormux.Cthulhu1.Service'
iface = busName.rsplit('.', 1)[0] + '.WineAccess' if args.extension_endpoint else busName
path = '/' + busName.replace('.', '/') + ('/WineAccess' if args.extension_endpoint else '')
readerArguments = ['--reader', args.reader] if args.reader else []
if args.extension_endpoint:
    readerArguments += ['--reader-path', path, '--reader-interface', iface]
capture_iface = 'org.stormux.Cthulhu1.WineReview'
capture_path = '/org/stormux/Cthulhu1/WineReview'

def interface_xml(review=True):
    methods = {'PresentAccessibleEvent': ['s']*7+['i','i','s'],
               'SpeakText': ['s','b'], 'CancelSpeech': []}
    if review:
        methods.update(RegisterWineReviewProvider=['o','u'], PresentWineReviewFocus=['s','s','s','t','s','s','b','(iiii)'])
    return '<node><interface name="'+iface+'">'+''.join(
        '<method name="'+name+'">'+''.join('<arg type="'+kind+'" direction="in"/>' for kind in kinds)+
        '<arg type="b" direction="out"/></method>' for name,kinds in methods.items())+'</interface></node>'

providers, focuses, events = {}, {}, []
def record(conn, sender, object_path, interface, method, params, invocation):
    data = params.unpack()
    accepted = True
    if method == 'RegisterWineReviewProvider':
        accepted = data == (capture_path, 1)
        if accepted: providers[sender] = data
    elif method == 'PresentWineReviewFocus':
        accepted = sender in providers
        if accepted: focuses[sender] = data
    else: events.append(data)
    invocation.return_value(GLib.Variant('(b)', (accepted,)))

connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
registration = connection.register_object(path, Gio.DBusNodeInfo.new_for_xml(interface_xml()).interfaces[0], record, None, None)
connection.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','RequestName',
                     GLib.Variant('(su)',(busName,0)),None,Gio.DBusCallFlags.NONE,1000,None)

def pump_until(predicate, seconds=5):
    end = time.monotonic()+seconds
    while not predicate() and time.monotonic()<end:
        while GLib.MainContext.default().iteration(False): pass
        time.sleep(.005)
    assert predicate(), f'private review protocol condition timed out; providers={providers}; helper={helper.poll()}; app={app.poll()}'

def request(conn, sender, focus):
    result = []
    started = time.monotonic()
    def finished(source, pending):
        try: result.append((source.call_finish(pending), time.monotonic()-started))
        except GLib.Error as error: result.append(error)
    conn.call(sender,capture_path,capture_iface,'CaptureWindow',GLib.Variant('(sst)',focus[1:4]),
              GLib.VariantType.new('(a{sv})'),Gio.DBusCallFlags.NONE,1500,None,finished)
    return result

with wine_test_directory(prefix='wine-review-wire-') as temporary:
    env = os.environ.copy()
    env.update(WINEPREFIX=str(Path(temporary)/'prefix'),WINEDEBUG=os.environ.get('WINE_ACCESS_TEST_DEBUG','-all'),WINEDLLOVERRIDES='mscoree,mshtml=',CTHULHU_WINE_ACCESS_MANAGED='1')
    fixture = build/('review-fixture32.exe' if args.is32 else 'review-fixture64.exe')
    helper, app, secondary = None, None, None
    def configure(mode):
        subprocess.run(['wine',str(fixture),'--configure',str(mode)],env=env,check=True,timeout=5)
        while GLib.MainContext.default().iteration(False): pass
    def capture():
        end=time.monotonic()+.15
        while time.monotonic()<end:
            while GLib.MainContext.default().iteration(False): pass
            time.sleep(.005)
        sender, focus = next((s,f) for s,f in focuses.items() if f[4]=='Wine review fixture')
        result = request(connection,sender,focus)
        pump_until(lambda: bool(result),2)
        assert not isinstance(result[0],GLib.Error),f'{result[0]}; helper={helper.poll()}'
        return result[0]
    try:
        subprocess.run(['wineboot','-u'],env=env,check=True,timeout=45)
        helper = subprocess.Popen([str(build/'wine-access.exe'),'--no-controller',*readerArguments],env=env)
        app = subprocess.Popen(['wine',str(fixture)],env=env)
        pump_until(lambda: any(f[4]=='Wine review fixture' for f in focuses.values()),8)
        reply, elapsed = capture()
        fields = reply.unpack()[0]
        assert fields['status']=='ok' and fields['version']==1, fields
        assert elapsed<.8 and fields['nodes']
        wire = reply.get_child_value(0)
        assert wire.lookup_value('version',GLib.VariantType.new('u')) is not None
        assert wire.lookup_value('revision',GLib.VariantType.new('t')) is not None
        assert wire.lookup_value('nodes',GLib.VariantType.new('aa{sv}')) is not None
        nodes=wire.lookup_value('nodes',GLib.VariantType.new('aa{sv}'))
        for index in range(nodes.n_children()):
            node=nodes.get_child_value(index)
            for key,kind in {'id':'s','parent':'s','role':'s','name':'s','protected':'b','visible':'b',
                             'text_known':'b','caret_known':'b','caret':'t','caret_utf16':'t',
                             'bounds':'(iiii)','ranges':'a(tttts(iiii))'}.items():
                assert node.lookup_value(key,GLib.VariantType.new(kind)) is not None,(key,kind)
        strings = str(fields)
        assert 'alpha beta' in strings and 'output second' in strings and 'native & item' in strings
        assert 'private-review-secret' not in strings and 'hidden-review-secret' not in strings
        subprocess.run(['wine',str(fixture),'--check'],env=env,check=True,timeout=5)
        print('PASS: test_capture_wire_contract')
        other = Gio.DBusConnection.new_for_address_sync(os.environ['DBUS_SESSION_BUS_ADDRESS'],
            Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT|Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION,None,None)
        # Owning the unselected reader name must not authorize capture.
        otherIface = 'org.stormux.Cthulhu1.Service' if args.reader == 'orca' else 'org.gnome.Orca1.Service'
        other.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','RequestName',
                        GLib.Variant('(su)',(otherIface,0)),None,Gio.DBusCallFlags.NONE,1000,None)
        sender,focus = next((s,f) for s,f in focuses.items() if f[4]=='Wine review fixture')
        wrong = request(other,sender,focus); pump_until(lambda:bool(wrong),2)
        assert isinstance(wrong[0],GLib.Error) and 'AccessDenied' in str(wrong[0])
        print('PASS: test_wrong_caller')
        # Move the selected name to a new connection. The helper must register
        # again, authorize the new owner, and immediately reject the old owner.
        other.register_object(path, Gio.DBusNodeInfo.new_for_xml(interface_xml()).interfaces[0],record,None,None)
        providers.clear(); focuses.clear()
        connection.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','ReleaseName',
                             GLib.Variant('(s)',(busName,)),None,Gio.DBusCallFlags.NONE,1000,None)
        other.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','RequestName',
                        GLib.Variant('(su)',(busName,0)),None,Gio.DBusCallFlags.NONE,1000,None)
        pump_until(lambda:sender in providers and sender in focuses)
        focus=focuses[sender]
        wrong=request(connection,sender,focus); pump_until(lambda:bool(wrong),2)
        assert isinstance(wrong[0],GLib.Error) and 'AccessDenied' in str(wrong[0])
        accepted=request(other,sender,focus); pump_until(lambda:bool(accepted),2)
        assert not isinstance(accepted[0],GLib.Error), accepted[0]
        assert accepted[0][0].unpack()[0]['status']=='ok'
        providers.clear(); focuses.clear()
        other.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','ReleaseName',
                        GLib.Variant('(s)',(busName,)),None,Gio.DBusCallFlags.NONE,1000,None)
        connection.call_sync('org.freedesktop.DBus','/org/freedesktop/DBus','org.freedesktop.DBus','RequestName',
                             GLib.Variant('(su)',(busName,0)),None,Gio.DBusCallFlags.NONE,1000,None)
        pump_until(lambda:sender in providers and sender in focuses)
        other.close_sync(None)
        print('PASS: test_reader_owner_replacement')
        configure(33)
        reply,_ = capture(); data=reply.unpack()[0]
        assert data['status']=='ok', data['status']
        ready=False
        for node in data['nodes']:
            assert '\0' not in node['name']
            for start,end,native_start,native_end,text,bounds in node['ranges']:
                assert '\0' not in text
                assert end-start==len(text)
                assert native_end-native_start==len(text.encode('utf-16-le'))//2
                if text=='Ready': ready=True
        assert ready, ('Padded MSAA status name lost its exposed text', data)
        subprocess.run(['wine',str(fixture),'--check'],env=env,check=True,timeout=5)
        print('PASS: test_padded_status_name_wire_lengths')
        configure(9)
        reply,_ = capture(); data=reply.unpack()[0]
        assert data['status'] in ('ok','stale')
        if data['status']=='stale': assert not data['nodes']
        else: assert 'omega zeta' in str(data) and 'alpha beta' not in str(data)
        print('PASS: test_same_length_capture_race')
        configure(21)
        reply,elapsed=capture(); data=reply.unpack()[0]
        assert data['status']=='unavailable' and not data['nodes'] and elapsed<.8,(data,elapsed)
        subprocess.run(['wine',str(fixture),'--check-remote'],env=env,check=True,timeout=5)
        print('PASS: test_native_buffer_lifetime_after_deadline')
        # Exiting the helper during a custom message must not invalidate the
        # application's remote allocation while its window procedure runs.
        configure(21)
        sender,focus=next((s,f) for s,f in focuses.items() if f[4]=='Wine review fixture')
        request(connection,sender,focus)
        end=time.monotonic()+.2
        while time.monotonic()<end:
            while GLib.MainContext.default().iteration(False): pass
            time.sleep(.005)
        helper.terminate(); helper.wait(timeout=3)
        subprocess.run(['wine',str(fixture),'--check-remote'],env=env,check=True,timeout=5)
        print('PASS: test_native_buffer_lifetime_after_helper_stop')
        helper=subprocess.Popen([str(build/'wine-access.exe'),'--no-controller',*readerArguments],env=env)
        pump_until(lambda:any(s != sender and f[4]=='Wine review fixture' for s,f in focuses.items()),5)
        focuses.pop(sender,None)
        configure(7)
        reply,_ = capture(); data=reply.unpack()[0]
        assert data['status']=='limited' and not data['nodes']
        configure(10)
        sender,focus = next((s,f) for s,f in focuses.items() if f[4]=='Wine review fixture')
        requests = [request(connection,sender,focus) for _ in range(5)]
        secondary = subprocess.Popen(['wine',str(fixture),'--secondary'],env=env)
        pump_until(lambda:all(requests),2)
        statuses = [result[0][0].unpack()[0]['status'] for result in requests]
        assert 'busy' in statuses, statuses
        assert 'unavailable' in statuses, statuses
        for result in requests:
            assert result[0][1]<.8, result[0][1]
            assert not result[0][0].unpack()[0]['nodes']
        print('PASS: test_busy_capture; test_capture_deadline')
        pump_until(lambda:any(event[9]=='Wine secondary fixture' and event[1]=='focus' for event in events),1)
        print('PASS: test_stalled_capture_keeps_focus_speech')
        started=time.monotonic(); helper.terminate(); helper.wait(timeout=3)
        assert time.monotonic()-started<3
        print('PASS: test_stop_during_capture')
        secondary.terminate(); secondary.wait(timeout=3)
        time.sleep(2)
        connection.unregister_object(registration)
        registration = connection.register_object(path,Gio.DBusNodeInfo.new_for_xml(interface_xml(False)).interfaces[0],record,None,None)
        events.clear()
        helper = subprocess.Popen([str(build/'wine-access.exe'),'--no-controller',*readerArguments],env=env)
        end=time.monotonic()+1
        while time.monotonic()<end:
            while GLib.MainContext.default().iteration(False): pass
            time.sleep(.005)
        configure(11)
        pump_until(lambda:bool(events),5)
        assert helper.poll() is None
        print('PASS: test_old_reader')
    finally:
        for process in (app,helper,secondary):
            if process is not None and process.poll() is None:
                process.terminate()
                try: process.wait(timeout=3)
                except subprocess.TimeoutExpired: process.kill(); process.wait(timeout=3)
        subprocess.run(['wineserver','-k'],env=env,timeout=5)