def quit_cb(loop):
loop.quit()
+ print('timed out while waiting')
def wait(millis):
global loop
+ before = GLib.get_monotonic_time()
loop = GLib.MainLoop()
GLib.timeout_add(millis, quit_cb, loop)
loop.run()
+ if verbose:
+ time = (GLib.get_monotonic_time() - before) / 1000
+ print(f'waited for {time} milliseconds')
display = None
window = None
if verbose:
print(f'got key press: {keyval}, state {state}')
- assert expected_change != None, "Unexpected key press"
- assert expected_change['type'] == 'press', "Key press event expected"
- assert keyval == expected_change['keyval'], "Unexpected keyval in key press event"
- assert state == expected_change['state'], "Unexpected state in key press event"
+ assert expected_change != None, 'Unexpected key press'
+ assert expected_change['type'] == 'press', 'Key press event expected'
+ assert keyval == expected_change['keyval'], 'Unexpected keyval in key press event'
+ assert state == expected_change['state'], 'Unexpected state in key press event'
expected_change = None
loop.quit()
if verbose:
print(f'got key release: {keyval}, state {state}')
- assert expected_change != None, "Unexpected key release"
- assert expected_change['type'] == 'release', "Key release event expected"
- assert keyval == expected_change['keyval'], "Unexpected keyval in key release event"
- assert state == expected_change['state'], "Unexpected state in key release event"
+ assert expected_change != None, 'Unexpected key release'
+ assert expected_change['type'] == 'release', 'Key release event expected'
+ assert keyval == expected_change['keyval'], 'Unexpected keyval in key release event'
+ assert state == expected_change['state'], 'Unexpected state in key release event'
expected_change = None
loop.quit()
if verbose:
print(f'got motion: {x}, {y}')
if expected_change != None:
- assert expected_change['type'] == 'motion', "Motion event expected"
- assert x == expected_change['x'], "Unexpected x coord in motion event"
- assert y == expected_change['y'], "Unexpected y coord in motion event"
+ assert expected_change['type'] == 'motion', 'Motion event expected'
+ assert x == expected_change['x'], 'Unexpected x coord in motion event'
+ assert y == expected_change['y'], 'Unexpected y coord in motion event'
expected_change = None
loop.quit()
if verbose:
print(f'got enter: {x}, {y}')
- assert expected_change != None, "Unexpected enter"
- assert expected_change['type'] == 'enter', "Enter event expected"
- assert x == expected_change['x'], "Unexpected x coord in enter event"
- assert y == expected_change['y'], "Unexpected y coord in enter event"
+ assert expected_change != None, 'Unexpected enter'
+ assert expected_change['type'] == 'enter', 'Enter event expected'
+ assert x == expected_change['x'], 'Unexpected x coord in enter event'
+ assert y == expected_change['y'], 'Unexpected y coord in enter event'
expected_change = None
loop.quit()
if verbose:
print(f'got pressed')
- assert expected_change != None, "Unexpected event"
- assert expected_change['type'] == 'press', "Button press expected"
- assert expected_change['button'] == controller.get_current_button(), "Unexpected button pressed"
- assert x == expected_change['x'], "Unexpected x coord in motion event"
- assert y == expected_change['y'], "Unexpected y coord in motion event"
+ assert expected_change != None, 'Unexpected event'
+ assert expected_change['type'] == 'press', 'Button press expected'
+ assert expected_change['button'] == controller.get_current_button(), 'Unexpected button pressed'
+ assert x == expected_change['x'], 'Unexpected x coord in motion event'
+ assert y == expected_change['y'], 'Unexpected y coord in motion event'
expected_change = None
loop.quit()
if verbose:
print(f'got released')
- assert expected_change != None, "Unexpected event"
- assert expected_change['type'] == 'release', "Button release expected"
+ assert expected_change != None, 'Unexpected event'
+ assert expected_change['type'] == 'release', 'Button release expected'
expected_change = None
loop.quit()
'state' : state
}
wait(timeout)
- assert expected_change == None, "Expected event did not happen"
+ assert expected_change == None, 'Expected event did not happen'
def expect_key_release(keyval, state, timeout):
global expected_change
'state' : state
}
wait(timeout)
- assert expected_change == None, "Expected event did not happen"
+ assert expected_change == None, 'Expected event did not happen'
def expect_motion(x, y, timeout):
global expected_change
'y' : y
}
wait(timeout)
- assert expected_change == None, "Expected event did not happen"
+ assert expected_change == None, 'Expected event did not happen'
def expect_enter(x, y, timeout):
global expected_change
'y' : y
}
wait(timeout)
- assert expected_change == None, "Expected event did not happen"
+ assert expected_change == None, 'Expected event did not happen'
def expect_button_press(button, x, y, timeout):
global expected_change
'y' : y
}
wait(timeout)
- assert expected_change == None, "Button press did not arrive"
+ assert expected_change == None, 'Button press did not arrive'
def expect_button_release(button, x, y, timeout):
global expected_change
'y' : y
}
wait(timeout)
- assert expected_change == None, "Button release did not arrive"
+ assert expected_change == None, 'Button release did not arrive'
def got_active(object, pspec):
global loop
wait(2000)
- assert window.is_active(), "Observer not active"
- assert window.get_width() == 1024, "Window not maximized"
- assert window.get_height() == 768, "Window not maximized"
+ assert window.is_active(), 'Observer not active'
+ assert window.get_width() == 1024, 'Window not maximized'
+ assert window.get_height() == 768, 'Window not maximized'
# we need to wait out the map animation, or pointer coords will be off
wait(1000)
def basic_keyboard_tests():
try:
+ if verbose:
+ print('Starting basic keyboard tests')
+
launch_observer()
key_press(Gdk.KEY_a)
stop_observer()
except AssertionError as e:
- print("Error in basic_keyboard_tests: {0}".format(e))
+ print(f'Error in basic_keyboard_tests: {e}')
terminate()
def basic_pointer_tests():
try:
+ if verbose:
+ print('Starting basic pointer tests')
+
pointer_move(-100.0, -100.0)
launch_observer()
stop_observer()
except AssertionError as e:
- print("Error in basic_pointer_tests: {0}".format(e))
+ print(f'Error in basic_pointer_tests: {e}')
terminate()
ds_window = None
global loop
if verbose:
- print(f'got drag begin')
- assert expected_change != None, "Unexpected drag begin"
- assert expected_change['type'] == 'drag', "Drag begin expected"
+ print('got drag begin')
+ assert expected_change != None, 'Unexpected drag begin'
+ assert expected_change['type'] == 'drag', 'Drag begin expected'
expected_change = None
loop.quit()
wait(2000)
- assert ds_window.is_active(), "drag source not active"
- assert ds_window.get_width() == 1024, "Window not maximized"
- assert ds_window.get_height() == 768, "Window not maximized"
+ assert ds_window.is_active(), 'drag source not active'
+ assert ds_window.get_width() == 1024, 'Window not maximized'
+ assert ds_window.get_height() == 768, 'Window not maximized'
# we need to wait out the map animation, or pointer coords will be off
wait(1000)
if verbose:
print(f'got drop {value}')
- assert expected_change != None, "Unexpected drop begin"
- assert expected_change['type'] == 'drop', "Drop expected"
- assert expected_change['value'] == value, "Unexpected value dropped"
+ assert expected_change != None, 'Unexpected drop begin'
+ assert expected_change['type'] == 'drop', 'Drop expected'
+ assert expected_change['value'] == value, 'Unexpected value dropped'
expected_change = None
loop.quit()
wait(2000)
- assert dt_window.is_active(), "drop target not active"
- assert dt_window.get_width() == 1024, "Window not maximized"
- assert dt_window.get_height() == 768, "Window not maximized"
+ assert dt_window.is_active(), 'drop target not active'
+ assert dt_window.get_width() == 1024, 'Window not maximized'
+ assert dt_window.get_height() == 768, 'Window not maximized'
# we need to wait out the map animation, or pointer coords will be off
wait(1000)
'type' : 'drag',
}
wait(timeout)
- assert expected_change == None, "DND operation not started"
+ assert expected_change == None, 'DND operation not started'
def expect_drop(value, timeout):
global expected_change
'value' : value
}
wait(timeout)
- assert expected_change == None, "Drop has not happened"
+ assert expected_change == None, 'Drop has not happened'
def dnd_tests():
try:
+ if verbose:
+ print('Starting dnd tests')
+
pointer_move(-100, -100)
launch_drag_source('abc')
stop_drop_target()
stop_drag_source()
except AssertionError as e:
- print("Error in dnd_tests: {0}".format(e))
+ print(f'Error in dnd_tests: {e}')
terminate()
def session_closed_cb():
print('Session closed')
+def run_commands():
+ basic_keyboard_tests()
+ basic_pointer_tests()
+ dnd_tests()
+
def mutter_appeared(name):
global remote_desktop
global session
global done
if verbose:
- print("mutter appeared on the bus")
+ print('mutter appeared on the bus')
remote_desktop = bus.get('org.gnome.Mutter.RemoteDesktop',
'/org/gnome/Mutter/RemoteDesktop')
device_types = remote_desktop.Get('org.gnome.Mutter.RemoteDesktop', 'SupportedDeviceTypes')
- assert device_types & 1 == 1, "No keyboard"
- assert device_types & 2 == 2, "No pointer"
+ assert device_types & 1 == 1, 'No keyboard'
+ assert device_types & 2 == 2, 'No pointer'
screen_cast = bus.get('org.gnome.Mutter.ScreenCast',
'/org/gnome/Mutter/ScreenCast')
key_release(Gdk.KEY_Control_L)
pointer_move(-100, -100)
- basic_keyboard_tests()
- basic_pointer_tests()
- dnd_tests()
+ run_commands()
session.Stop()
+ if verbose:
+ print('Done running commands, exiting...')
done = True
def mutter_vanished():
global done
if remote_desktop != None:
if verbose:
- print("mutter left the bus")
+ print('mutter left the bus')
done = True
bus = SessionBus()
def stream_added_closure(name):
def stream_added(node_id):
+ if verbose:
+ print('pipewire stream added')
+
monitor = monitors[name];
freq = monitor['freq'];
# Use gstreamer out-of-process, since the gst gl support gets
# itself into a twist with its wayland connection when monitors
# disappear
- pipeline_desc = f'gst-launch-1.0 --verbose pipewiresrc path={node_id} ! video/x-raw,max-framerate={freq}/1,width={width},height={height} ! videoconvert ! glimagesink'
+ pipeline_desc = f'gst-launch-1.0 pipewiresrc path={node_id} ! video/x-raw,max-framerate={freq}/1,width={width},height={height} ! videoconvert ! glimagesink' # >& gstreamer-monitor.log'
if verbose:
print(f'launching {pipeline_desc}')
monitor['pipeline'] = subprocess.Popen([pipeline_desc], shell=True)
session_path = screen_cast.CreateSession({})
session = bus.get('org.gnome.Mutter.ScreenCast', session_path)
monitors[name] = {
- "session": session,
- "width": width,
- "height": height,
- "scale": scale,
- "freq": freq
+ 'session': session,
+ 'width': width,
+ 'height': height,
+ 'scale': scale,
+ 'freq': freq
}
stream_path = session.RecordVirtual({})
stream = bus.get('org.gnome.Mutter.ScreenCast', stream_path)
session = monitor['session']
session.Stop()
except KeyError:
- print("failed to remove monitor")
+ print('failed to remove monitor')
monitors[name] = None
expected_change = None
def quit_cb(loop):
loop.quit()
+ print('timed out while waiting')
def wait(millis):
global loop
+ before = GLib.get_monotonic_time()
loop = GLib.MainLoop()
GLib.timeout_add(millis, quit_cb, loop)
loop.run()
+ if verbose:
+ time = (GLib.get_monotonic_time() - before) / 1000
+ print(f'waited for {time} milliseconds')
def monitors_changed(monitors, position, removed, added):
global expected_change
- assert expected_change != None, "No change expected"
- assert position == expected_change['position'], "Unexpected position in monitors-changed"
- assert removed == expected_change['removed'], "Unexpected removed in monitors-changed"
- assert added == expected_change['added'], "Unexpected added in monitors-changed"
+ assert expected_change != None, 'No change expected'
+ assert position == expected_change['position'], 'Unexpected position in monitors-changed'
+ assert removed == expected_change['removed'], 'Unexpected removed in monitors-changed'
+ assert added == expected_change['added'], 'Unexpected added in monitors-changed'
if verbose:
- print('got expected change')
+ print('got expected monitors-changed signal')
expected_change = None
loop.quit()
print('launch observer')
monitor_model = display.get_monitors()
- assert monitor_model.get_n_items() == 0, "Unexpected initial monitors"
+ assert monitor_model.get_n_items() == 0, 'Unexpected initial monitors'
monitor_model.connect('items-changed', monitors_changed)
def expect_monitors_changed(position, removed, added, timeout):
'added' : added
}
wait(timeout)
- assert expected_change == None, "Expected change did not happen"
+ assert expected_change == None, 'Expected change did not happen'
def got_connector(monitor, pspec):
loop.quit()
assert monitor_model.get_n_items() > position, f'Monitor {position} not present'
monitor = monitor_model.get_item(position)
if monitor.get_connector() == None:
+ if verbose:
+ print('waiting for connector')
handler = monitor.connect('notify::connector', got_connector)
wait(500)
monitor.disconnect(handler)
- assert monitor.is_valid(), "Monitor is not valid"
+ assert monitor.get_connector() != None, 'Monitor has no connector'
+ assert monitor.is_valid(), 'Monitor is not valid'
geometry = monitor.get_geometry()
- assert geometry.width == width, "Unexpected monitor width"
- assert geometry.height == height, "Unexpected monitor height"
- assert monitor.get_scale_factor() == scale, "Unexpected scale factor"
- assert monitor.get_refresh_rate() == freq, "Unexpected monitor frequency"
+ assert geometry.width == width, 'Unexpected monitor width'
+ assert geometry.height == height, 'Unexpected monitor height'
+ assert monitor.get_scale_factor() == scale, 'Unexpected scale factor'
+ assert monitor.get_refresh_rate() == freq, 'Unexpected monitor frequency'
if verbose:
print(f'monitor {position}: {geometry.width}x{geometry.height} frequency {monitor.get_refresh_rate()} scale {monitor.get_scale_factor()} model \'{monitor.get_model()}\' connector \'{monitor.get_connector()}\'')
try:
launch_observer()
- add_monitor("0", width=100, height=100, scale=1, freq=60)
- expect_monitors_changed(0, 0, 1, 5000)
+ add_monitor('0', width=100, height=100, scale=1, freq=60)
+ expect_monitors_changed(0, 0, 1, 10000)
expect_monitor (position=0, width=100, height=100, scale=1, freq=60000)
- add_monitor("1", width=1024, height=768, scale=1, freq=144)
- expect_monitors_changed(1, 0, 1, 5000)
+ add_monitor('1', width=1024, height=768, scale=1, freq=144)
+ expect_monitors_changed(1, 0, 1, 10000)
expect_monitor (position=1, width=1024, height=768, scale=1, freq=144000)
- remove_monitor("0")
+ remove_monitor('0')
expect_monitors_changed(0, 1, 0, 11000) # mutter takes 10 seconds to remove it
- remove_monitor("1")
+ remove_monitor('1')
expect_monitors_changed(0, 1, 0, 11000)
except AssertionError as e:
- print("Error: {0}".format(e))
+ print(f'Error: {e}')
terminate()
def mutter_appeared(name):
global screen_cast
global done
if verbose:
- print("mutter appeared on the bus")
+ print('mutter appeared on the bus')
screen_cast = bus.get('org.gnome.Mutter.ScreenCast',
'/org/gnome/Mutter/ScreenCast')
run_commands()
if verbose:
- print ("Done running commands, exiting...")
+ print ('Done running commands, exiting...')
done = True
def mutter_vanished():
global done
if screen_cast != None:
if verbose:
- print("mutter left the bus")
+ print('mutter left the bus')
done = True
bus = SessionBus()