From be62216e340c6fe3f0e192473bf89cb3f122e11a Mon Sep 17 00:00:00 2001 From: Jonny Lamb Date: Thu, 31 May 2012 15:42:46 +0100 Subject: tube tests: remove assertions about Tubes channels Signed-off-by: Jonny Lamb --- tests/twisted/avahi/tubes/request-muc-tubes.py | 36 ++++----- tests/twisted/avahi/tubes/two-muc-dbus-tubes.py | 94 ++++++++--------------- tests/twisted/avahi/tubes/two-muc-stream-tubes.py | 56 ++++---------- 3 files changed, 63 insertions(+), 123 deletions(-) (limited to 'tests') diff --git a/tests/twisted/avahi/tubes/request-muc-tubes.py b/tests/twisted/avahi/tubes/request-muc-tubes.py index 6ac74bd6..f5921c0c 100644 --- a/tests/twisted/avahi/tubes/request-muc-tubes.py +++ b/tests/twisted/avahi/tubes/request-muc-tubes.py @@ -44,8 +44,8 @@ def test(q, bus, conn): EventPattern('dbus-signal', signal='NewChannel'), EventPattern('dbus-signal', signal='NewChannels'), ) - path2 = ret.value[0] - chan = wrap_channel(bus.get_object(conn.bus_name, path2), + tube_path = ret.value[0] + chan = wrap_channel(bus.get_object(conn.bus_name, tube_path), 'StreamTube') handle = conn.RequestHandles(HT_ROOM, ['my-second-room'])[0] @@ -61,19 +61,16 @@ def test(q, bus, conn): # text and tube channels are announced channels = new_sig.args[0] - assert len(channels) == 3 - got_text, got_tubes, got_tube = False, False, False + assert len(channels) == 2 + got_text, got_tube = False, False for path, props in channels: if props[CHANNEL_TYPE] == CHANNEL_TYPE_TEXT: got_text = True assert props[REQUESTED] == False - elif props[CHANNEL_TYPE] == CHANNEL_TYPE_TUBES: - got_tubes = True - assert props[REQUESTED] == False elif props[CHANNEL_TYPE] == CHANNEL_TYPE_STREAM_TUBE: got_tube = True - assert path == path2 + assert path == tube_path assert props == tube_props else: assert False @@ -85,25 +82,22 @@ def test(q, bus, conn): assert props[INITIATOR_ID] == self_name # ensure the same channel - yours, ensured_path, ensured_props = conn.Requests.EnsureChannel( - { CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE, - TARGET_HANDLE_TYPE: HT_ROOM, - TARGET_HANDLE: handle, - STREAM_TUBE_SERVICE: 'loldongs', - }) - - # TODO: get rid of Tubes and this'll start making more sense - return +# yours, ensured_path, ensured_props = conn.Requests.EnsureChannel( +# { CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE, +# TARGET_HANDLE_TYPE: HT_ROOM, +# TARGET_HANDLE: handle, +# STREAM_TUBE_SERVICE: 'loldongs', +# }) - assert not yours - assert ensured_path == path2, (ensured_path, path2) +# assert not yours +# assert ensured_path == tube_path, (ensured_path, tube_path2) conn.Disconnect() q.expect_many( EventPattern('dbus-signal', signal='Closed', - path=path2), - EventPattern('dbus-signal', signal='ChannelClosed', args=[path2]), + path=tube_path), + EventPattern('dbus-signal', signal='ChannelClosed', args=[tube_path]), EventPattern('dbus-signal', signal='StatusChanged', args=[2, 1]), ) diff --git a/tests/twisted/avahi/tubes/two-muc-dbus-tubes.py b/tests/twisted/avahi/tubes/two-muc-dbus-tubes.py index 9fd17ca3..c34a8e2e 100644 --- a/tests/twisted/avahi/tubes/two-muc-dbus-tubes.py +++ b/tests/twisted/avahi/tubes/two-muc-dbus-tubes.py @@ -71,40 +71,26 @@ def test(q, bus, conn): e = q.expect('dbus-signal', signal='NewChannels') channels = e.args[0] - assert len(channels) == 2 + assert len(channels) == 1 # get the list of all channels to check that newly announced ones are in it all_channels = conn.Get(cs.CONN_IFACE_REQUESTS, 'Channels', dbus_interface=cs.PROPERTIES_IFACE, byte_arrays=True) - got_tubes, got_tube = False, False - for path, props in channels: - if props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_TUBES: - got_tubes = True - assert props[cs.REQUESTED] == False - assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP] - elif props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_DBUS_TUBE: - got_tube = True - assert props[cs.REQUESTED] == True - assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP, - cs.CHANNEL_IFACE_TUBE] - assert props[cs.DBUS_TUBE_SERVICE_NAME] == 'com.example.TestCase' - assert props[cs.DBUS_TUBE_SUPPORTED_ACCESS_CONTROLS] == [ - cs.SOCKET_ACCESS_CONTROL_CREDENTIALS, cs.SOCKET_ACCESS_CONTROL_LOCALHOST] - - contact1_tube = wrap_channel(bus.get_object(conn.bus_name, path), 'DBusTube') - tube1_path = path - else: - assert False - - assert props[cs.INITIATOR_HANDLE] == conn1_self_handle - assert props[cs.INITIATOR_ID] == contact1_name - assert props[cs.TARGET_ID] == muc_name - - assert (path, props) in all_channels, (path, props) - - assert got_tubes - assert got_tube + path, props = channels[0] + + assert props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_DBUS_TUBE + assert props[cs.REQUESTED] == True + assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP, + cs.CHANNEL_IFACE_TUBE] + assert props[cs.DBUS_TUBE_SERVICE_NAME] == 'com.example.TestCase' + assert props[cs.DBUS_TUBE_SUPPORTED_ACCESS_CONTROLS] == [ + cs.SOCKET_ACCESS_CONTROL_CREDENTIALS, cs.SOCKET_ACCESS_CONTROL_LOCALHOST] + + contact1_tube = wrap_channel(bus.get_object(conn.bus_name, path), 'DBusTube') + tube1_path = path + + assert (path, props) in all_channels, (path, props) state = contact1_tube.Properties.Get(cs.CHANNEL_IFACE_TUBE, 'State') assert state == cs.TUBE_CHANNEL_STATE_NOT_OFFERED @@ -132,41 +118,27 @@ def test(q, bus, conn): EventPattern('dbus-signal', signal='DBusNamesChanged', interface=cs.CHANNEL_TYPE_DBUS_TUBE)) channels = e.args[0] - assert len(channels) == 2 + assert len(channels) == 1 # get the list of all channels to check that newly announced ones are in it all_channels = conn2.Get(cs.CONN_IFACE_REQUESTS, 'Channels', dbus_interface=cs.PROPERTIES_IFACE, byte_arrays=True) - got_tubes, got_tube = False, False - for path, props in channels: - if props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_TUBES: - got_tubes = True - assert props[cs.REQUESTED] == False - assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP] - elif props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_DBUS_TUBE: - got_tube = True - assert props[cs.REQUESTED] == False - assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP, - cs.CHANNEL_IFACE_TUBE] - assert props[cs.TUBE_PARAMETERS] == sample_parameters - assert props[cs.DBUS_TUBE_SERVICE_NAME] == 'com.example.TestCase' - assert props[cs.DBUS_TUBE_SUPPORTED_ACCESS_CONTROLS] == [ - cs.SOCKET_ACCESS_CONTROL_CREDENTIALS, cs.SOCKET_ACCESS_CONTROL_LOCALHOST] - - contact2_tube = wrap_channel(bus.get_object(conn.bus_name, path), 'DBusTube') - tube2_path = path - else: - assert False - - assert props[cs.INITIATOR_HANDLE] == contact1_handle_on_conn2 - assert props[cs.INITIATOR_ID] == contact1_name - assert props[cs.TARGET_ID] == muc_name - - assert (path, props) in all_channels, (path, props) - - assert got_tubes - assert got_tube + path, props = channels[0] + + assert props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_DBUS_TUBE + assert props[cs.REQUESTED] == False + assert props[cs.INTERFACES] == [cs.CHANNEL_IFACE_GROUP, + cs.CHANNEL_IFACE_TUBE] + assert props[cs.TUBE_PARAMETERS] == sample_parameters + assert props[cs.DBUS_TUBE_SERVICE_NAME] == 'com.example.TestCase' + assert props[cs.DBUS_TUBE_SUPPORTED_ACCESS_CONTROLS] == [ + cs.SOCKET_ACCESS_CONTROL_CREDENTIALS, cs.SOCKET_ACCESS_CONTROL_LOCALHOST] + + contact2_tube = wrap_channel(bus.get_object(conn.bus_name, path), 'DBusTube') + tube2_path = path + + assert (path, props) in all_channels, (path, props) # second connection: check DBusNamesChanged signal assert dbus_names_e.path == tube2_path @@ -234,10 +206,9 @@ def test(q, bus, conn): q.expect('tube-dbus-return', method='MyMethod', value=[420]) call_async(q, contact1_tube, 'Close') - _, _, _, _, dbus_names_e = q.expect_many( + _, _, _, dbus_names_e = q.expect_many( EventPattern('dbus-return', method='Close'), EventPattern('dbus-signal', signal='Closed'), - EventPattern('dbus-signal', signal='TubeClosed'), EventPattern('dbus-signal', signal='ChannelClosed'), EventPattern('dbus-signal', signal='DBusNamesChanged', interface=cs.CHANNEL_TYPE_DBUS_TUBE, path=tube2_path)) @@ -253,7 +224,6 @@ def test(q, bus, conn): q.expect_many( EventPattern('dbus-return', method='Close'), EventPattern('dbus-signal', signal='Closed'), - EventPattern('dbus-signal', signal='TubeClosed'), EventPattern('dbus-signal', signal='ChannelClosed')) conn.Disconnect() diff --git a/tests/twisted/avahi/tubes/two-muc-stream-tubes.py b/tests/twisted/avahi/tubes/two-muc-stream-tubes.py index 50913668..51a7f880 100644 --- a/tests/twisted/avahi/tubes/two-muc-stream-tubes.py +++ b/tests/twisted/avahi/tubes/two-muc-stream-tubes.py @@ -22,7 +22,6 @@ sample_parameters = dbus.Dictionary({ test_string = "This string travels on a tube !" muc_name = "test-two-muc-stream-tubes" -muc2_name = "test-two-muc-stream-tubes-2" SERVER_WELCOME_MSG = "Welcome!" @@ -78,28 +77,24 @@ def test(q, bus, conn): conn.Requests.CreateChannel({ CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE, TARGET_HANDLE_TYPE: HT_ROOM, - TARGET_ID: muc2_name, + TARGET_ID: muc_name, STREAM_TUBE_SERVICE: 'test'}) e = q.expect('dbus-signal', signal='NewChannels', - predicate=lambda e: len(e.args[0]) == 3) + predicate=lambda e: len(e.args[0]) == 2) channels = e.args[0] # get the list of all channels to check that newly announced ones are in it all_channels = conn.Properties.Get(CONN_IFACE_REQUESTS, 'Channels', byte_arrays=True) - got_text, got_tubes, got_tube = False, False, False + got_text, got_tube = False, False for path, props in channels: if props[CHANNEL_TYPE] == CHANNEL_TYPE_TEXT: got_text = True assert props[REQUESTED] == False text1 = wrap_channel(bus.get_object(conn.bus_name, path), 'Text') txt_path = path - elif props[CHANNEL_TYPE] == CHANNEL_TYPE_TUBES: - got_tubes = True - assert props[REQUESTED] == False - assert props[INTERFACES] == [CHANNEL_IFACE_GROUP] elif props[CHANNEL_TYPE] == CHANNEL_TYPE_STREAM_TUBE: got_tube = True assert props[REQUESTED] == True @@ -114,12 +109,11 @@ def test(q, bus, conn): assert props[INITIATOR_HANDLE] == conn1_self_handle assert props[INITIATOR_ID] == contact1_name - assert props[TARGET_ID] == muc2_name + assert props[TARGET_ID] == muc_name assert (path, props) in all_channels, (path, props) assert got_text - assert got_tubes assert got_tube state = contact1_tube.Properties.Get(CHANNEL_IFACE_TUBE, 'State') @@ -146,40 +140,24 @@ def test(q, bus, conn): # tubes channel is created e = q.expect('dbus-signal', signal='NewChannels') channels = e.args[0] - assert len(channels) == 2 + assert len(channels) == 1 # get the list of all channels to check that newly announced ones are in it all_channels = conn2.Properties.Get(CONN_IFACE_REQUESTS, 'Channels', byte_arrays=True) - got_tubes, got_tube = False, False - for path, props in channels: - if props[CHANNEL_TYPE] == CHANNEL_TYPE_TUBES: - got_tubes = True - assert props[REQUESTED] == False - assert props[INTERFACES] == [CHANNEL_IFACE_GROUP] - elif props[CHANNEL_TYPE] == CHANNEL_TYPE_STREAM_TUBE: - got_tube = True - assert props[REQUESTED] == False - assert props[INTERFACES] == [CHANNEL_IFACE_GROUP, - CHANNEL_IFACE_TUBE] - assert props[STREAM_TUBE_SERVICE] == 'test' - assert props[TUBE_PARAMETERS] == sample_parameters - - contact2_tube = wrap_channel(bus.get_object(conn.bus_name, path), - 'StreamTube') - tube2_path = path - else: - assert False + path, props = channels[0] + assert props[REQUESTED] == False + assert props[INTERFACES] == [CHANNEL_IFACE_GROUP, + CHANNEL_IFACE_TUBE] + assert props[STREAM_TUBE_SERVICE] == 'test' + assert props[TUBE_PARAMETERS] == sample_parameters - assert props[INITIATOR_HANDLE] == contact1_handle_on_conn2 - assert props[INITIATOR_ID] == contact1_name - assert props[TARGET_ID] == muc2_name + assert (path, props) in all_channels, (path, props) - assert (path, props) in all_channels, (path, props) - - assert got_tubes - assert got_tube + contact2_tube = wrap_channel(bus.get_object(conn.bus_name, path), + 'StreamTube') + tube2_path = path state = contact2_tube.Properties.Get(CHANNEL_IFACE_TUBE, 'State') assert state == TUBE_CHANNEL_STATE_LOCAL_PENDING @@ -230,12 +208,11 @@ def test(q, bus, conn): assert client_received.data == string.swapcase(test_string) call_async(q, contact1_tube, 'Close') - _, e1, e2, _, _, _ = q.expect_many( + _, e1, e2, _, _ = q.expect_many( EventPattern('dbus-return', method='Close'), EventPattern('dbus-signal', signal='ConnectionClosed', path=tube1_path), EventPattern('dbus-signal', signal='ConnectionClosed', path=tube2_path), EventPattern('dbus-signal', signal='Closed'), - EventPattern('dbus-signal', signal='TubeClosed'), EventPattern('dbus-signal', signal='ChannelClosed')) conn_id, error, dbus_msg = e1.args @@ -250,7 +227,6 @@ def test(q, bus, conn): q.expect_many( EventPattern('dbus-return', method='Close'), EventPattern('dbus-signal', signal='Closed'), - EventPattern('dbus-signal', signal='TubeClosed'), EventPattern('dbus-signal', signal='ChannelClosed')) conn.Disconnect() -- cgit v1.2.1