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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
"""
Test text channel initiated by me, using Requests.EnsureChannel
"""
import dbus
from twisted.words.xish import domish
from hazetest import exec_test
from servicetest import call_async, EventPattern, unwrap
import pprint
import constants as cs
def test(q, bus, conn, stream):
self_handle = conn.Properties.Get(cs.CONN, "SelfHandle")
jids = ['foo@bar.com', 'truc@cafe.fr']
handles = conn.get_contact_handles_sync(jids)
properties = conn.GetAll(
'org.freedesktop.Telepathy.Connection.Interface.Requests',
dbus_interface=dbus.PROPERTIES_IFACE)
# Difference from Gabble: Haze's roster channels spring to life even if you
# haven't received the XMPP roster.
text_channels = [c for c in properties.get('Channels')
if c[1][cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_TEXT
]
assert text_channels == [], text_channels
assert ({'org.freedesktop.Telepathy.Channel.ChannelType':
'org.freedesktop.Telepathy.Channel.Type.Text',
'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
},
['org.freedesktop.Telepathy.Channel.TargetHandle',
'org.freedesktop.Telepathy.Channel.TargetID'
],
) in properties.get('RequestableChannelClasses'),\
properties['RequestableChannelClasses']
test_ensure_ensure(q, conn, self_handle, jids[0], handles[0])
test_request_ensure(q, conn, self_handle, jids[1], handles[1])
conn.Disconnect()
q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
def test_ensure_ensure(q, conn, self_handle, jid, handle):
"""
Test ensuring a non-existant channel twice. The first call should succeed
with Yours=True; the subsequent call should succeed with Yours=False
"""
# Check that Ensuring a channel that doesn't exist succeeds
call_async(q, conn.Requests, 'EnsureChannel', request_props (handle))
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='EnsureChannel'),
EventPattern('dbus-signal', signal='NewChannel'),
EventPattern('dbus-signal', signal='NewChannels'),
)
assert len(ret.value) == 3
yours, path, emitted_props = ret.value
# The channel was created in response to the call, and we were the only
# requestor, so we should get Yours=True
assert yours, ret.value
check_props(emitted_props, self_handle, handle, jid)
assert len(old_sig.args) == 5
old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
assert old_path == path
assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
# check that handle type == contact handle
assert old_ht == 1
assert old_h == handle
assert old_sh == True # suppress handler
assert len(new_sig.args) == 1
assert len(new_sig.args[0]) == 1 # one channel
assert len(new_sig.args[0][0]) == 2 # two struct members
assert new_sig.args[0][0][0] == path
assert new_sig.args[0][0][1] == emitted_props
properties = conn.GetAll(
'org.freedesktop.Telepathy.Connection.Interface.Requests',
dbus_interface=dbus.PROPERTIES_IFACE)
assert new_sig.args[0][0] in properties['Channels'], \
(new_sig.args[0][0], properties['Channels'])
# Now try Ensuring a channel which already exists
call_async(q, conn.Requests, 'EnsureChannel', request_props (handle))
ret_ = q.expect('dbus-return', method='EnsureChannel')
assert len(ret_.value) == 3
yours_, path_, emitted_props_ = ret_.value
# Someone's already responsible for this channel, so we should get
# Yours=False
assert not yours_, ret_.value
assert path == path_, (path, path_)
assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
def test_request_ensure(q, conn, self_handle, jid, handle):
"""
Test Creating a non-existant channel, then Ensuring the same channel.
The call to Ensure should succeed with Yours=False.
"""
call_async(q, conn.Requests, 'CreateChannel', request_props (handle))
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='CreateChannel'),
EventPattern('dbus-signal', signal='NewChannel'),
EventPattern('dbus-signal', signal='NewChannels'),
)
assert len(ret.value) == 2
path, emitted_props = ret.value
check_props(emitted_props, self_handle, handle, jid)
assert len(old_sig.args) == 5
old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
assert old_path == path
assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
# check that handle type == contact handle
assert old_ht == 1
assert old_h == handle
assert old_sh == True # suppress handler
assert len(new_sig.args) == 1
assert len(new_sig.args[0]) == 1 # one channel
assert len(new_sig.args[0][0]) == 2 # two struct members
assert new_sig.args[0][0][0] == path
assert new_sig.args[0][0][1] == emitted_props
properties = conn.GetAll(
'org.freedesktop.Telepathy.Connection.Interface.Requests',
dbus_interface=dbus.PROPERTIES_IFACE)
assert new_sig.args[0][0] in properties['Channels'], \
(new_sig.args[0][0], properties['Channels'])
# Now try Ensuring that same channel.
call_async(q, conn.Requests, 'EnsureChannel', request_props (handle))
ret_ = q.expect('dbus-return', method='EnsureChannel')
assert len(ret_.value) == 3
yours_, path_, emitted_props_ = ret_.value
# Someone's already responsible for this channel, so we should get
# Yours=False
assert not yours_, ret_.value
assert path == path_, (path, path_)
assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
def check_props(props, self_handle, handle, jid):
assert props['org.freedesktop.Telepathy.Channel.ChannelType'] ==\
'org.freedesktop.Telepathy.Channel.Type.Text'
assert props['org.freedesktop.Telepathy.Channel.'
'TargetHandleType'] == 1
assert props['org.freedesktop.Telepathy.Channel.TargetHandle'] ==\
handle
assert props['org.freedesktop.Telepathy.Channel.TargetID'] == jid
assert props['org.freedesktop.Telepathy.Channel.'
'Requested'] == True
assert props['org.freedesktop.Telepathy.Channel.'
'InitiatorHandle'] == self_handle
assert props['org.freedesktop.Telepathy.Channel.'
'InitiatorID'] == 'test@localhost'
def request_props(handle):
return { 'org.freedesktop.Telepathy.Channel.ChannelType':
'org.freedesktop.Telepathy.Channel.Type.Text',
'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
'org.freedesktop.Telepathy.Channel.TargetHandle': handle,
}
if __name__ == '__main__':
exec_test(test)
|
