tests: Add get_secondary_bus_and_device helper

It returns a new bus object and the device on that bus. This allows
testing what happens if a client disconnects from the bus.
This commit is contained in:
Benjamin Berg
2020-12-16 13:31:44 +01:00
parent 7c9a04c2ae
commit 6ae4f5e939

View File

@ -462,6 +462,26 @@ class FPrintdVirtualDeviceBaseTest(FPrintdTest):
return (enrolled, enroll_map) return (enrolled, enroll_map)
def get_secondary_bus_and_device(self):
addr = os.environ['DBUS_SYSTEM_BUS_ADDRESS']
# Get a separat bus connection
bus = Gio.DBusConnection.new_for_address_sync(addr,
Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION |
Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT, None, None)
assert bus.is_closed() == False
dev_path = self.device.get_object_path()
dev = Gio.DBusProxy.new_sync(bus,
Gio.DBusProxyFlags.DO_NOT_AUTO_START,
None,
'net.reactivated.Fprint',
dev_path,
'net.reactivated.Fprint.Device',
None)
return bus, dev
class FPrintdManagerTests(FPrintdVirtualDeviceBaseTest): class FPrintdManagerTests(FPrintdVirtualDeviceBaseTest):
@ -700,23 +720,7 @@ class FPrintdVirtualDeviceTest(FPrintdVirtualDeviceBaseTest):
self.device.Release() self.device.Release()
def test_claim_disconnect(self): def test_claim_disconnect(self):
addr = os.environ['DBUS_SYSTEM_BUS_ADDRESS'] bus, dev = self.get_secondary_bus_and_device()
# Get a separat bus connection
dbus = Gio.DBusConnection.new_for_address_sync(addr,
Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION |
Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT, None, None)
assert dbus.is_closed() == False
dev_path = self.device.get_object_path()
dev = Gio.DBusProxy.new_sync(dbus,
Gio.DBusProxyFlags.DO_NOT_AUTO_START,
None,
'net.reactivated.Fprint',
dev_path,
'net.reactivated.Fprint.Device',
None)
def call_done(obj, result, user_data): def call_done(obj, result, user_data):
# Ignore the callback (should be an error) # Ignore the callback (should be an error)
@ -726,8 +730,8 @@ class FPrintdVirtualDeviceTest(FPrintdVirtualDeviceBaseTest):
dev.Claim('(s)', 'testuser', result_handler=call_done) dev.Claim('(s)', 'testuser', result_handler=call_done)
# Ensure the call is on the wire, then close immediately # Ensure the call is on the wire, then close immediately
dbus.flush_sync() bus.flush_sync()
dbus.close_sync() bus.close_sync()
time.sleep(1) time.sleep(1)