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
|
"""
Test connecting to a IRC channel via the Requests interface
"""
import functools
from idletest import exec_test, BaseIRCServer, sync_stream
from servicetest import (
EventPattern, call_async, sync_dbus, make_channel_proxy, assertEquals,
assertSameSets, assertContains,
)
import constants as cs
import dbus
class DelayJoinServer(BaseIRCServer):
def handleJOIN(self, args):
# do nothing; wait for the test to call sendJoin().
return
def build_request(conn, channel_name, use_room):
rccs = conn.Properties.Get(cs.CONN_IFACE_REQUESTS,
'RequestableChannelClasses')
if use_room:
# We allow TargetHandleType in Room-flavoured requests, but it has to
# be None if specified.
assertContains(
({ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT },
[cs.TARGET_HANDLE_TYPE, cs.ROOM_NAME],
), rccs)
request = {
cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.ROOM_NAME: '#idletest'
}
else:
assertContains(
({ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
},
[cs.TARGET_HANDLE, cs.TARGET_ID]
), rccs)
request = {
cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT,
cs.TARGET_HANDLE_TYPE: cs.HT_ROOM,
cs.TARGET_ID: '#idletest',
}
return dbus.Dictionary(request, signature='sv')
def test(q, bus, conn, stream, use_room=False):
conn.Connect()
q.expect_many(
EventPattern('dbus-signal', signal='StatusChanged', args=[1, 1]),
EventPattern('irc-connected'))
q.expect('dbus-signal', signal='SelfHandleChanged')
q.expect('dbus-signal', signal='StatusChanged', args=[0, 1])
self_handle = conn.GetSelfHandle()
request = build_request(conn, '#idletest', use_room)
call_async(q, conn.Requests, 'CreateChannel', request)
# Idle should try to join the channel.
q.expect('stream-JOIN')
# Meanwhile, in another application...
call_async(q, conn, 'EnsureChannel', request,
dbus_interface=cs.CONN_IFACE_REQUESTS)
sync_dbus(bus, q, conn)
# Now the ircd responds:
stream.sendJoin('#idletest')
cc, ec = q.expect_many(
EventPattern('dbus-return', method='CreateChannel'),
EventPattern('dbus-return', method='EnsureChannel'),
)
nc = q.expect('dbus-signal', signal='NewChannels')
path, props = cc.value
assert props[cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_TEXT
assertSameSets(
[cs.CHANNEL_IFACE_GROUP,
cs.CHANNEL_IFACE_PASSWORD,
cs.CHANNEL_IFACE_MESSAGES,
cs.CHANNEL_IFACE_ROOM,
cs.CHANNEL_IFACE_SUBJECT,
cs.CHANNEL_IFACE_ROOM_CONFIG,
cs.CHANNEL_IFACE_DESTROYABLE,
], props[cs.INTERFACES])
assert props[cs.TARGET_HANDLE_TYPE] == cs.HT_ROOM
assert props[cs.TARGET_ID] == '#idletest'
assertEquals('#idletest', props[cs.ROOM_NAME])
assertEquals('', props[cs.ROOM_SERVER])
assert props[cs.TARGET_HANDLE] == \
conn.RequestHandles(cs.HT_ROOM, ['#idletest'])[0]
assert props[cs.REQUESTED]
assert props[cs.INITIATOR_HANDLE] == self_handle
assert props[cs.INITIATOR_ID] == \
conn.InspectHandles(cs.HT_CONTACT, [self_handle])[0]
ec_yours, ec_path, ec_props = ec.value
assert not ec_yours
assert ec_path == path
assert ec_props == props
channels = nc.args[0]
assert len(channels) == 1
nc_path, nc_props = channels[0]
assert nc_path == path
assert nc_props == props
# And again?
ec_ = conn.EnsureChannel(request,
dbus_interface=cs.CONN_IFACE_REQUESTS)
assert ec.value == ec_
chans = conn.Get(cs.CONN_IFACE_REQUESTS, 'Channels',
dbus_interface=cs.PROPERTIES_IFACE)
assert len(chans) == 1
assert chans[0] == (path, props)
chan = make_channel_proxy(conn, path, 'Channel')
# Make sure Close()ing the channel makes it respawn. This avoids the old
# bug where empathy-chat crashing booted you out of all your channels.
patterns = [EventPattern('stream-PART')]
q.forbid_events(patterns)
chan.Close()
q.expect('dbus-signal', signal='Closed', path=chan.object_path)
e = q.expect('dbus-signal', signal='NewChannels')
path, props = e.args[0][0]
assertEquals(chan.object_path, path)
# We requested the channel originally, but we didn't request it popping
# back up.
assertEquals(0, props[cs.INITIATOR_HANDLE])
assert not props[cs.REQUESTED]
# Check that ensuring a respawned channel does what you'd expect.
ec_yours, ec_path, ec_props = conn.EnsureChannel(request,
dbus_interface=cs.CONN_IFACE_REQUESTS)
assert not ec_yours
assertEquals(chan.object_path, ec_path)
assertEquals(props, ec_props)
sync_stream(q, stream)
q.unforbid_events(patterns)
chan.RemoveMembers([self_handle], "bye bye cruel\r\nworld",
dbus_interface=cs.CHANNEL_IFACE_GROUP)
part_event = q.expect('stream-PART')
# This is a regression test for
# <https://bugs.freedesktop.org/show_bug.cgi?id=34812>, where part messages
# were not correctly colon-quoted.
#
# It is also a regression test for
# <https://bugs.freedesktop.org/show_bug.cgi?id=34840>, where newlines
# weren't stripped from part messages. We check that both \r and \n are
# replaced by harmless spaces.
assertEquals("bye bye cruel world", part_event.data[1])
stream.sendPart('#idletest', stream.nick)
q.expect('dbus-signal', signal='Closed')
chans = conn.Get(cs.CONN_IFACE_REQUESTS, 'Channels',
dbus_interface=cs.PROPERTIES_IFACE)
assert len(chans) == 0
if __name__ == '__main__':
exec_test(test, protocol=DelayJoinServer)
exec_test(functools.partial(test, use_room=True), protocol=DelayJoinServer)
|
