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
151
152
153
154
155
156
157
|
"""
Test requesting of muc text channels using the old and new request API.
"""
import dbus
import avahitest
from twisted.words.xish import domish
from saluttest import exec_test, wait_for_contact_list
from servicetest import call_async, EventPattern, \
tp_name_prefix, tp_path_prefix, make_channel_proxy
CHANNEL_TYPE_TEXT = 'org.freedesktop.Telepathy.Channel.Type.Text'
HT_ROOM = 2
def test(q, bus, conn):
self_name = 'testsuite' + '@' + avahitest.get_host_name()
conn.Connect()
q.expect('dbus-signal', signal='StatusChanged', args=[0L, 0L])
# FIXME: this is a hack to be sure to have all the contact list channels
# announced so they won't interfere with the muc ones announces.
wait_for_contact_list(q, conn)
# check if we can request roomlist channels
properties = conn.GetAll(
tp_name_prefix + '.Connection.Interface.Requests',
dbus_interface='org.freedesktop.DBus.Properties')
assert ({tp_name_prefix + '.Channel.ChannelType':
CHANNEL_TYPE_TEXT,
tp_name_prefix + '.Channel.TargetHandleType': HT_ROOM,
},
[tp_name_prefix + '.Channel.TargetHandle',
tp_name_prefix + '.Channel.TargetID'],
) in properties.get('RequestableChannelClasses'),\
properties['RequestableChannelClasses']
# request a muc channel using the old API
handle = conn.RequestHandles(HT_ROOM, ['my-first-room'])[0]
call_async(q, conn, 'RequestChannel', CHANNEL_TYPE_TEXT, HT_ROOM, handle, True)
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='RequestChannel'),
EventPattern('dbus-signal', signal='NewChannel'),
EventPattern('dbus-signal', signal='NewChannels'),
)
path1 = ret.value[0]
chan = make_channel_proxy(conn, path1, "Channel")
assert new_sig.args[0][0][0] == path1
props = new_sig.args[0][0][1]
assert props[tp_name_prefix + '.Channel.ChannelType'] ==\
CHANNEL_TYPE_TEXT
assert props[tp_name_prefix + '.Channel.TargetHandleType'] == HT_ROOM
assert props[tp_name_prefix + '.Channel.TargetHandle'] == handle
assert props[tp_name_prefix + '.Channel.TargetID'] == 'my-first-room'
assert props[tp_name_prefix + '.Channel.Requested'] == True
assert props[tp_name_prefix + '.Channel.InitiatorHandle'] \
== conn.GetSelfHandle()
assert props[tp_name_prefix + '.Channel.InitiatorID'] \
== self_name
assert old_sig.args[0] == path1
assert old_sig.args[1] == CHANNEL_TYPE_TEXT
assert old_sig.args[2] == HT_ROOM # handle type
assert old_sig.args[3] == handle # handle
# Exercise basic Channel Properties from spec 0.17.7
channel_props = chan.GetAll(
tp_name_prefix + '.Channel',
dbus_interface='org.freedesktop.DBus.Properties')
assert channel_props.get('TargetHandle') == handle,\
channel_props.get('TargetHandle')
assert channel_props['TargetID'] == 'my-first-room', channel_props
assert channel_props.get('TargetHandleType') == HT_ROOM,\
channel_props.get('TargetHandleType')
assert channel_props.get('ChannelType') == \
CHANNEL_TYPE_TEXT, channel_props.get('ChannelType')
assert channel_props['Requested'] == True
assert channel_props['InitiatorID'] == self_name
assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
requestotron = dbus.Interface(conn,
tp_name_prefix + '.Connection.Interface.Requests')
# create muc channel using new API
call_async(q, requestotron, 'CreateChannel',
{ tp_name_prefix + '.Channel.ChannelType':
CHANNEL_TYPE_TEXT,
tp_name_prefix + '.Channel.TargetHandleType': HT_ROOM,
tp_name_prefix + '.Channel.TargetID': 'my-second-room',
})
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='CreateChannel'),
EventPattern('dbus-signal', signal='NewChannel'),
EventPattern('dbus-signal', signal='NewChannels'),
)
path2 = ret.value[0]
chan = make_channel_proxy(conn, path2, "Channel")
handle = conn.RequestHandles(HT_ROOM, ['my-second-room'])[0]
props = ret.value[1]
assert props[tp_name_prefix + '.Channel.ChannelType'] ==\
CHANNEL_TYPE_TEXT
assert props[tp_name_prefix + '.Channel.TargetHandleType'] == HT_ROOM
assert props[tp_name_prefix + '.Channel.TargetHandle'] == handle
assert props[tp_name_prefix + '.Channel.TargetID'] == 'my-second-room'
assert props[tp_name_prefix + '.Channel.Requested'] == True
assert props[tp_name_prefix + '.Channel.InitiatorHandle'] \
== conn.GetSelfHandle()
assert props[tp_name_prefix + '.Channel.InitiatorID'] \
== self_name
assert new_sig.args[0][0][0] == path2
assert new_sig.args[0][0][1] == props
assert old_sig.args[0] == path2
assert old_sig.args[1] == CHANNEL_TYPE_TEXT
assert old_sig.args[2] == HT_ROOM # handle type
assert old_sig.args[3] == handle # handle
assert old_sig.args[4] == True # suppress handler
# ensure roomlist channel
yours, ensured_path, ensured_props = ret.value = requestotron.EnsureChannel(
{ tp_name_prefix + '.Channel.ChannelType':
CHANNEL_TYPE_TEXT,
tp_name_prefix + '.Channel.TargetHandleType': HT_ROOM,
tp_name_prefix + '.Channel.TargetHandle': handle,
})
assert not yours
assert ensured_path == path2, (ensured_path, path2)
conn.Disconnect()
q.expect_many(
EventPattern('dbus-signal', signal='Closed',
path=path1),
EventPattern('dbus-signal', signal='Closed',
path=path2),
EventPattern('dbus-signal', signal='ChannelClosed', args=[path1]),
EventPattern('dbus-signal', signal='ChannelClosed', args=[path2]),
EventPattern('dbus-signal', signal='StatusChanged', args=[2, 1]),
)
if __name__ == '__main__':
exec_test(test)
|
