238 lines
14 KiB
Python
238 lines
14 KiB
Python
"""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)
|