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
|
"""
Test text channel initiated by me, using Requests.EnsureChannel
"""
import dbus
from gabbletest import exec_test
from servicetest import call_async, EventPattern
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(
cs.CONN_IFACE_REQUESTS, dbus_interface=cs.PROPERTIES_IFACE)
assert properties.get('Channels') == [], properties['Channels']
assert ({cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
},
[cs.TARGET_HANDLE, cs.TARGET_ID],
) 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])
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, new_sig = q.expect_many(
EventPattern('dbus-return', method='EnsureChannel'),
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(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(
cs.CONN_IFACE_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, new_sig = q.expect_many(
EventPattern('dbus-return', method='CreateChannel'),
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(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(
cs.CONN_IFACE_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[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_TEXT
assert props[cs.TARGET_HANDLE_TYPE] == cs.HT_CONTACT
assert props[cs.TARGET_HANDLE] == handle
assert props[cs.TARGET_ID] == jid
assert props[cs.REQUESTED] == True
assert props[cs.INITIATOR_HANDLE] == self_handle
assert props[cs.INITIATOR_ID] == 'test@localhost'
def request_props(handle):
return { cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
cs.TARGET_HANDLE: handle,
}
if __name__ == '__main__':
exec_test(test)
|
