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
|
"""
A smoketest for the XMPP console API.
"""
from servicetest import (
ProxyWrapper, EventPattern,
call_async, assertEquals, assertContains, assertNotEquals, sync_dbus,
)
from gabbletest import exec_test, acknowledge_iq, elem, elem_iq
from config import PLUGINS_ENABLED
from twisted.words.xish import domish
import ns
import constants as cs
CONSOLE_PLUGIN_IFACE = "org.freedesktop.Telepathy.Gabble.Plugin.Console"
STACY = 'stacy@pilgrim.lit'
if not PLUGINS_ENABLED:
print "NOTE: built without --enable-plugins, not testing XMPP console"
raise SystemExit(77)
def send_unrecognised_get(q, stream):
stream.send(
elem_iq(stream, 'get')(
elem('urn:unimaginative', 'dont-handle-me-bro')
))
return q.expect('stream-iq', iq_type='error')
def test(q, bus, conn, stream):
rccs = conn.Properties.Get(cs.CONN_IFACE_REQUESTS,
'RequestableChannelClasses')
fixed = {
cs.CHANNEL_TYPE: CONSOLE_PLUGIN_IFACE,
cs.TARGET_HANDLE_TYPE: cs.HT_NONE,
}
allowed = []
assertContains((fixed, allowed), rccs)
path, _ = conn.Requests.CreateChannel({ cs.CHANNEL_TYPE: CONSOLE_PLUGIN_IFACE })
console = ProxyWrapper(bus.get_object(conn.bus_name, path),
CONSOLE_PLUGIN_IFACE,
{'Channel': cs.CHANNEL})
assert not console.Properties.Get(CONSOLE_PLUGIN_IFACE, 'SpewStanzas')
es = [
EventPattern('dbus-signal', signal='StanzaReceived'),
EventPattern('dbus-signal', signal='StanzaSent'),
]
q.forbid_events(es)
call_async(q, console, 'SendIQ', 'get', STACY,
'<coffee xmlns="urn:unimaginative"/>')
e = q.expect('stream-iq', iq_type='get', query_ns='urn:unimaginative',
query_name='coffee')
acknowledge_iq(stream, e.stanza)
e = q.expect('dbus-return', method='SendIQ')
type_, body = e.value
assertEquals('result', type_)
# We just assume the body works.
# Turn on signalling incoming and outgoing stanzas
console.Properties.Set(CONSOLE_PLUGIN_IFACE, 'SpewStanzas', True)
sync_dbus(bus, q, conn)
q.unforbid_events(es)
send_unrecognised_get(q, stream)
e = q.expect('dbus-signal', signal='StanzaReceived')
xml, = e.args
assertContains('<iq', xml)
assertContains('<dont-handle-me-bro', xml)
signal = q.expect('dbus-signal', signal='StanzaSent')
assertContains('service-unavailable', signal.args[0])
# Turn off spewing out stanzas; check it works.
console.Properties.Set(CONSOLE_PLUGIN_IFACE, 'SpewStanzas', False)
q.forbid_events(es)
send_unrecognised_get(q, stream)
sync_dbus(bus, q, conn)
# Try sending just any old stanza
console.SendStanza('''
<message to='%(stacy)s' type='headline'>
<body>
Hi sis.
</body>
</message>''' % { 'stacy': STACY })
e = q.expect('stream-message', to=STACY, message_type='headline')
# Make sure that Wocky has filled in the jabber:client namespace we
# carelessly omitted.
message = e.stanza
assertEquals('message', message.name)
assertEquals(ns.CLIENT, message.uri)
body = message.firstChildElement()
assertEquals('body', body.name)
assertEquals(ns.CLIENT, body.uri)
console.Channel.Close()
if __name__ == '__main__':
exec_test(test)
|
