/usr/lib/telepathy-gabble-tests/twisted/muc/test-ensure.py is in telepathy-gabble-tests 0.18.3-2build1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """
Test that EnsureChannel works for MUCs, particularly in the case when there
are several pending requests for the same MUC.
"""
from gabbletest import make_result_iq, exec_test, make_muc_presence
from servicetest import (call_async, EventPattern, assertContains,
assertEquals)
import constants as cs
def test(q, bus, conn, stream):
# Need to call this asynchronously as it involves Gabble sending us a
# query.
jids = ['chat@conf.localhost', 'chien@conf.localhost']
call_async(q, conn, 'RequestHandles', 2, jids)
event = q.expect('dbus-return', method='RequestHandles')
room_handles = event.value[0]
test_create_ensure(q, conn, bus, stream, jids[0], room_handles[0])
test_ensure_ensure(q, conn, bus, stream, jids[1], room_handles[1])
def test_create_ensure(q, conn, bus, stream, room_jid, room_handle):
# Call both Create and Ensure for the same channel.
call_async(q, conn.Requests, 'CreateChannel',
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
cs.TARGET_HANDLE: room_handle,
})
call_async(q, conn.Requests, 'EnsureChannel',
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
cs.TARGET_HANDLE: room_handle,
})
mc, _ = q.expect_many(
EventPattern('dbus-signal', signal='MembersChanged'),
EventPattern('stream-presence', to=('%s/test' % room_jid)))
msg, added, removed, local_pending, remote_pending, actor, reason = mc.args
assert added == [], mc.args
assert removed == [], mc.args
assert local_pending == [], mc.args
assert len(remote_pending) == 1, mc.args
# Send presence for other member of room.
stream.send(make_muc_presence('owner', 'moderator', room_jid, 'bob'))
# Send presence for own membership of room.
stream.send(make_muc_presence('none', 'participant', room_jid, 'test'))
mc = q.expect('dbus-signal', signal='MembersChanged')
msg, added, removed, local_pending, remote_pending, actor, reason = mc.args
assert len(added) == 2, mc.args
assert removed == [], mc.args
assert local_pending == [], mc.args
assert remote_pending == [], mc.args
members = conn.InspectHandles(1, added)
members.sort()
assert members == ['%s/bob' % room_jid, '%s/test' % room_jid], members
create_event, ensure_event = q.expect_many(
EventPattern('dbus-return', method='CreateChannel'),
EventPattern('dbus-return', method='EnsureChannel'))
assert len(create_event.value) == 2
c_path, c_props = create_event.value
assert len(ensure_event.value) == 3
yours, e_path, e_props = ensure_event.value
assert c_path == e_path, (c_path, e_path)
assert c_props == e_props, (c_props, e_props)
assert not yours
assertContains('text/plain', c_props[cs.SUPPORTED_CONTENT_TYPES])
assertEquals(0, c_props[cs.MESSAGE_PART_SUPPORT_FLAGS])
assertEquals(
cs.DELIVERY_REPORTING_SUPPORT_FLAGS_RECEIVE_FAILURES |
cs.DELIVERY_REPORTING_SUPPORT_FLAGS_RECEIVE_SUCCESSES,
c_props[cs.DELIVERY_REPORTING_SUPPORT])
def test_ensure_ensure(q, conn, bus, stream, room_jid, room_handle):
# Call Ensure twice for the same channel.
call_async(q, conn.Requests, 'EnsureChannel',
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
cs.TARGET_HANDLE: room_handle,
})
call_async(q, conn.Requests, 'EnsureChannel',
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
cs.TARGET_HANDLE: room_handle,
})
mc, _ = q.expect_many(
EventPattern('dbus-signal', signal='MembersChanged'),
EventPattern('stream-presence', to=('%s/test' % room_jid)))
msg, added, removed, local_pending, remote_pending, actor, reason = mc.args
assert added == [], mc.args
assert removed == [], mc.args
assert local_pending == [], mc.args
assert len(remote_pending) == 1, mc.args
# Send presence for other member of room.
stream.send(make_muc_presence('owner', 'moderator', room_jid, 'bob'))
# Send presence for own membership of room.
stream.send(make_muc_presence('none', 'participant', room_jid, 'test'))
mc = q.expect('dbus-signal', signal='MembersChanged')
msg, added, removed, local_pending, remote_pending, actor, reason = mc.args
assert len(added) == 2, mc.args
assert removed == [], mc.args
assert local_pending == [], mc.args
assert remote_pending == [], mc.args
members = conn.InspectHandles(1, added)
members.sort()
assert members == ['%s/bob' % room_jid, '%s/test' % room_jid], members
# We should get two EnsureChannel returns
es = []
while len(es) < 2:
e = q.expect('dbus-return', method='EnsureChannel')
es.append(e)
e1, e2 = es
assert len(e1.value) == 3
yours1, path1, props1 = e1.value
assert len(e2.value) == 3
yours2, path2, props2 = e2.value
# Exactly one Ensure should get Yours=True.
assert (yours1 == (not yours2))
assert path1 == path2, (path1, path2)
assert props1 == props2, (props1, props2)
if __name__ == '__main__':
exec_test(test)
|