← Back to branch summary

~ubuntu-branches/ubuntu/precise/telepathy-mission-control-5/precise

« back to all changes in this revision

Viewing changes to test/twisted/dispatcher/dispatch-rejected-by-mini-plugin.py

Tags: upstream-5.5.0
Import upstream version 5.5.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/twisted/dispatcher/dispatch-rejected-by-mini-plugin.py
 
1
# Copyright (C) 2009 Nokia Corporation
 
2
# Copyright (C) 2009 Collabora Ltd.
 
3
#
 
4
# This library is free software; you can redistribute it and/or
 
5
# modify it under the terms of the GNU Lesser General Public
 
6
# License as published by the Free Software Foundation; either
 
7
# version 2.1 of the License, or (at your option) any later version.
 
8
#
 
9
# This library is distributed in the hope that it will be useful, but
 
10
# WITHOUT ANY WARRANTY; without even the implied warranty of
 
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 
12
# Lesser General Public License for more details.
 
13
#
 
14
# You should have received a copy of the GNU Lesser General Public
 
15
# License along with this library; if not, write to the Free Software
 
16
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 
17
# 02110-1301 USA
 
18
 
 
19
import dbus
 
20
"""Regression test for plugins rejecting an incoming channel immediately.
 
21
"""
 
22
 
 
23
import dbus
 
24
import dbus.service
 
25
 
 
26
from servicetest import EventPattern, tp_name_prefix, tp_path_prefix, \
 
27
        call_async, sync_dbus
 
28
from mctest import exec_test, SimulatedConnection, SimulatedClient, \
 
29
        create_fakecm_account, enable_fakecm_account, SimulatedChannel, \
 
30
        expect_client_setup
 
31
import constants as cs
 
32
 
 
33
def test(q, bus, mc):
 
34
    params = dbus.Dictionary({"account": "someguy@example.com",
 
35
        "password": "secrecy"}, signature='sv')
 
36
    cm_name_ref, account = create_fakecm_account(q, bus, mc, params)
 
37
    conn = enable_fakecm_account(q, bus, mc, account, params)
 
38
 
 
39
    text_fixed_properties = dbus.Dictionary({
 
40
        cs.CHANNEL + '.TargetHandleType': cs.HT_CONTACT,
 
41
        cs.CHANNEL + '.ChannelType': cs.CHANNEL_TYPE_TEXT,
 
42
        }, signature='sv')
 
43
 
 
44
    # Throughout this entire test, we should never be asked to approve or
 
45
    # handle a channel.
 
46
    forbidden = [
 
47
            EventPattern('dbus-method-call', method='AddDispatchOperation'),
 
48
            EventPattern('dbus-method-call', method='HandleChannels'),
 
49
            ]
 
50
    q.forbid_events(forbidden)
 
51
 
 
52
    # Two clients want to observe, approve and handle channels
 
53
    empathy = SimulatedClient(q, bus, 'Empathy',
 
54
            observe=[text_fixed_properties], approve=[text_fixed_properties],
 
55
            handle=[text_fixed_properties], bypass_approval=False)
 
56
    kopete = SimulatedClient(q, bus, 'Kopete',
 
57
            observe=[text_fixed_properties], approve=[text_fixed_properties],
 
58
            handle=[text_fixed_properties], bypass_approval=False)
 
59
 
 
60
    # wait for MC to download the properties
 
61
    expect_client_setup(q, [empathy, kopete])
 
62
 
 
63
    # subscribe to the OperationList interface (MC assumes that until this
 
64
    # property has been retrieved once, nobody cares)
 
65
 
 
66
    cd = bus.get_object(cs.CD, cs.CD_PATH)
 
67
    cd_props = dbus.Interface(cd, cs.PROPERTIES_IFACE)
 
68
    assert cd_props.Get(cs.CD_IFACE_OP_LIST, 'DispatchOperations') == []
 
69
 
 
70
    # This ID is special-cased by the mcp-plugin plugin, which rejects
 
71
    # channels to or from it by destroying them, without waiting for observers
 
72
    # to return
 
73
    target = 'rick.astley@example.net'
 
74
    channel_properties = dbus.Dictionary(text_fixed_properties,
 
75
            signature='sv')
 
76
    channel_properties[cs.CHANNEL + '.TargetID'] = target
 
77
    channel_properties[cs.CHANNEL + '.TargetHandle'] = \
 
78
            conn.ensure_handle(cs.HT_CONTACT, target)
 
79
    channel_properties[cs.CHANNEL + '.InitiatorID'] = target
 
80
    channel_properties[cs.CHANNEL + '.InitiatorHandle'] = \
 
81
            conn.ensure_handle(cs.HT_CONTACT, target)
 
82
    channel_properties[cs.CHANNEL + '.Requested'] = False
 
83
    channel_properties[cs.CHANNEL + '.Interfaces'] = \
 
84
            dbus.Array([cs.CHANNEL_IFACE_DESTROYABLE, cs.CHANNEL_IFACE_GROUP,
 
85
                ],signature='s')
 
86
 
 
87
    chan = SimulatedChannel(conn, channel_properties, group=True)
 
88
    chan.announce()
 
89
 
 
90
    # A channel dispatch operation is created
 
91
 
 
92
    e = q.expect('dbus-signal',
 
93
            path=cs.CD_PATH,
 
94
            interface=cs.CD_IFACE_OP_LIST,
 
95
            signal='NewDispatchOperation')
 
96
 
 
97
    cdo_path = e.args[0]
 
98
    cdo_properties = e.args[1]
 
99
 
 
100
    assert cdo_properties[cs.CDO + '.Account'] == account.object_path
 
101
    assert cdo_properties[cs.CDO + '.Connection'] == conn.object_path
 
102
    assert cs.CDO + '.Interfaces' in cdo_properties
 
103
 
 
104
    handlers = cdo_properties[cs.CDO + '.PossibleHandlers'][:]
 
105
    handlers.sort()
 
106
    assert handlers == [cs.tp_name_prefix + '.Client.Empathy',
 
107
            cs.tp_name_prefix + '.Client.Kopete'], handlers
 
108
 
 
109
    # The plugin realises we've been rickrolled, and responds. It calls Destroy
 
110
    # even though neither Empathy nor Kopete has returned from ObserveChannels
 
111
    # yet
 
112
    destruction, e, k = q.expect_many(
 
113
            EventPattern('dbus-method-call',
 
114
                path=chan.object_path,
 
115
                interface=cs.CHANNEL_IFACE_DESTROYABLE, method='Destroy',
 
116
                args=[], handled=False),
 
117
            EventPattern('dbus-method-call',
 
118
                path=empathy.object_path,
 
119
                interface=cs.OBSERVER, method='ObserveChannels',
 
120
                handled=False),
 
121
            EventPattern('dbus-method-call',
 
122
                path=kopete.object_path,
 
123
                interface=cs.OBSERVER, method='ObserveChannels',
 
124
                handled=False),
 
125
            )
 
126
    # treat the destruction like Close
 
127
    chan.Close(destruction)
 
128
 
 
129
    # Both Observers indicate that they are ready to proceed (somewhat late)
 
130
    q.dbus_return(k.message, signature='')
 
131
    q.dbus_return(e.message, signature='')
 
132
 
 
133
    # When the Observers have returned, the CDO finishes
 
134
    q.expect_many(
 
135
            EventPattern('dbus-signal', path=cdo_path,
 
136
                interface=cs.CDO, signal='Finished'),
 
137
            EventPattern('dbus-signal', path=cs.CD_PATH,
 
138
                interface=cs.CD_IFACE_OP_LIST,
 
139
                signal='DispatchOperationFinished',
 
140
                args=[cdo_path]),
 
141
            )
 
142
 
 
143
    # This ID is also special-cased
 
144
    target = 'mc.hammer@example.net'
 
145
    channel_properties = dbus.Dictionary(text_fixed_properties,
 
146
            signature='sv')
 
147
    channel_properties[cs.CHANNEL + '.TargetID'] = target
 
148
    channel_properties[cs.CHANNEL + '.TargetHandle'] = \
 
149
            conn.ensure_handle(cs.HT_CONTACT, target)
 
150
    channel_properties[cs.CHANNEL + '.InitiatorID'] = target
 
151
    channel_properties[cs.CHANNEL + '.InitiatorHandle'] = \
 
152
            conn.ensure_handle(cs.HT_CONTACT, target)
 
153
    channel_properties[cs.CHANNEL + '.Requested'] = False
 
154
    channel_properties[cs.CHANNEL + '.Interfaces'] = \
 
155
            dbus.Array([cs.CHANNEL_IFACE_DESTROYABLE, cs.CHANNEL_IFACE_GROUP,
 
156
                ],signature='s')
 
157
 
 
158
    chan = SimulatedChannel(conn, channel_properties, group=True)
 
159
    chan.announce()
 
160
 
 
161
    # A channel dispatch operation is created
 
162
 
 
163
    e = q.expect('dbus-signal',
 
164
            path=cs.CD_PATH,
 
165
            interface=cs.CD_IFACE_OP_LIST,
 
166
            signal='NewDispatchOperation')
 
167
 
 
168
    cdo_path = e.args[0]
 
169
    cdo_properties = e.args[1]
 
170
 
 
171
    assert cdo_properties[cs.CDO + '.Account'] == account.object_path
 
172
    assert cdo_properties[cs.CDO + '.Connection'] == conn.object_path
 
173
    assert cs.CDO + '.Interfaces' in cdo_properties
 
174
 
 
175
    handlers = cdo_properties[cs.CDO + '.PossibleHandlers'][:]
 
176
    handlers.sort()
 
177
    assert handlers == [cs.tp_name_prefix + '.Client.Empathy',
 
178
            cs.tp_name_prefix + '.Client.Kopete'], handlers
 
179
 
 
180
    # The plugin realises it's MC Hammer, and responds, but its response waits
 
181
    # for the observers to return
 
182
    e, k = q.expect_many(
 
183
            EventPattern('dbus-method-call',
 
184
                path=empathy.object_path,
 
185
                interface=cs.OBSERVER, method='ObserveChannels',
 
186
                handled=False),
 
187
            EventPattern('dbus-method-call',
 
188
                path=kopete.object_path,
 
189
                interface=cs.OBSERVER, method='ObserveChannels',
 
190
                handled=False),
 
191
            )
 
192
 
 
193
    sync_dbus(bus, q, account)
 
194
 
 
195
    # Both Observers indicate that they are ready to proceed
 
196
    q.dbus_return(k.message, signature='')
 
197
    q.dbus_return(e.message, signature='')
 
198
 
 
199
    _, _, e = q.expect_many(
 
200
            EventPattern('dbus-signal', path=cdo_path,
 
201
                interface=cs.CDO, signal='Finished'),
 
202
            EventPattern('dbus-signal', path=cs.CD_PATH,
 
203
                interface=cs.CD_IFACE_OP_LIST,
 
204
                signal='DispatchOperationFinished',
 
205
                args=[cdo_path]),
 
206
            EventPattern('dbus-method-call',
 
207
                path=chan.object_path,
 
208
                interface=cs.CHANNEL_IFACE_GROUP,
 
209
                method='RemoveMembersWithReason', args=[[conn.self_handle],
 
210
                    "Can't touch this", cs.GROUP_REASON_PERMISSION_DENIED],
 
211
                handled=False),
 
212
            )
 
213
    q.dbus_return(e.message, signature='')
 
214
    chan.close()
 
215
 
 
216
if __name__ == '__main__':
 
217
    exec_test(test, {})