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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
"""
Test tubes capabilities with Connection.Interface.ContactCapabilities.DRAFT
1. Check if Salut advertise the OOB caps
2. Receive presence and caps from contacts and check that
GetContactAttributes works correctly and that ContactCapabilitiesChanged is
correctly received.
- capa announced with FT
- capa announced without FT
- no capabilites announced (assume FT is supported)
"""
import dbus
from avahitest import AvahiAnnouncer, AvahiListener
from avahitest import get_host_name
from avahitest import txt_get_key
import avahi
from twisted.words.xish import xpath
from servicetest import EventPattern, assertContains, assertDoesNotContain
from saluttest import exec_test, make_result_iq
from xmppstream import setup_stream_listener
import ns
import constants as cs
from caps_helper import compute_caps_hash, ft_fixed_properties, \
ft_allowed_properties, ft_allowed_properties_with_metadata
from config import PACKAGE_STRING
ft_caps = (ft_fixed_properties, ft_allowed_properties)
ft_metadata_caps = (ft_fixed_properties, ft_allowed_properties_with_metadata)
# last value of the "ver" key we resolved. We use it to be sure that the
# modified caps has already be announced.
old_ver = ''
def receive_presence_and_ask_caps(q, stream, service, contact_name):
global old_ver
event_avahi = q.expect('service-resolved', service=service)
ver = txt_get_key(event_avahi.txt, "ver")
while ver == old_ver:
# be sure that the announced caps actually changes
event_avahi = q.expect('service-resolved', service=service)
ver = txt_get_key(event_avahi.txt, "ver")
old_ver = ver
hash = txt_get_key(event_avahi.txt, "hash")
node = txt_get_key(event_avahi.txt, "node")
assert hash == 'sha-1'
# ask caps
# FIXME: this only works because Salut ignores @to
request = """
<iq from='""" + contact_name + """'
id='receive-presence-and-ask-caps'
to='salut@jabber.org/resource'
type='get'>
<query xmlns='http://jabber.org/protocol/disco#info'
node='""" + node + '#' + ver + """'/>
</iq>
"""
stream.send(request)
# receive caps
event = q.expect('stream-iq', iq_type='result',
to=contact_name, iq_id='receive-presence-and-ask-caps',
query_ns='http://jabber.org/protocol/disco#info')
caps_str = str(xpath.queryForNodes('/iq/query/feature', event.stanza))
features = []
for feature in xpath.queryForNodes('/iq/query/feature', event.stanza):
features.append(feature['var'])
# Check if the hash matches the announced capabilities
assert ver == compute_caps_hash(['client/pc//%s' % PACKAGE_STRING], features, {})
assert ns.X_OOB in features
assert ns.IQ_OOB in features
def caps_contain(event, cap):
node = xpath.queryForNodes('/iq/query/feature[@var="%s"]'
% cap,
event.stanza)
if node is None:
return False
if len(node) != 1:
return False
var = node[0].attributes['var']
if var is None:
return False
return var == cap
def test_ft_caps_from_contact(q, bus, conn, client):
conn_caps_iface = dbus.Interface(conn, cs.CONN_IFACE_CONTACT_CAPS)
conn_contacts_iface = dbus.Interface(conn, cs.CONN_IFACE_CONTACTS)
# send presence with FT capa
ver = compute_caps_hash([], [ns.IQ_OOB], {})
txt_record = { "txtvers": "1", "status": "avail",
"node": client, "ver": ver, "hash": "sha-1"}
contact_name = "test-caps-ft@" + get_host_name()
listener, port = setup_stream_listener(q, contact_name)
announcer = AvahiAnnouncer(contact_name, "_presence._tcp", port,
txt_record)
# this is the first presence, Salut connects to the contact
e = q.expect('incoming-connection', listener = listener)
incoming = e.connection
# Salut looks up our capabilities
event = q.expect('stream-iq', connection = incoming,
query_ns='http://jabber.org/protocol/disco#info')
query_node = xpath.queryForNodes('/iq/query', event.stanza)[0]
assert query_node.attributes['node'] == \
client + '#' + ver, (query_node.attributes['node'], client, ver)
contact_handle = conn_contacts_iface.GetContactByID(contact_name, [])[0]
# send good reply
result = make_result_iq(event.stanza)
query = result.firstChildElement()
query['node'] = client + '#' + ver
feature = query.addElement('feature')
feature['var'] = ns.IQ_OOB
incoming.send(result)
# FT capa is announced
e = q.expect('dbus-signal', signal='ContactCapabilitiesChanged',
predicate=lambda e: contact_handle in e.args[0])
caps = e.args[0][contact_handle]
assertContains(ft_caps, caps)
# check the Contacts interface give the same caps
caps_via_contacts_iface = conn_contacts_iface.GetContactAttributes(
[contact_handle], [cs.CONN_IFACE_CONTACT_CAPS]) \
[contact_handle][cs.CONN_IFACE_CONTACT_CAPS + '/capabilities']
assert caps_via_contacts_iface == caps, caps_via_contacts_iface
# check if Salut announces the OOB capa
self_handle = conn.Properties.Get(cs.CONN, "SelfHandle")
self_handle_name = conn.Properties.Get(cs.CONN, "SelfID")
AvahiListener(q).listen_for_service("_presence._tcp")
e = q.expect('service-added', name = self_handle_name,
protocol = avahi.PROTO_INET)
service = e.service
service.resolve()
receive_presence_and_ask_caps(q, incoming, service, contact_name)
# capa announced without FT
ver = compute_caps_hash([], ["http://telepathy.freedesktop.org/xmpp/pony"], {})
txt_record = { "txtvers": "1", "status": "avail",
"node": client, "ver": ver, "hash": "sha-1"}
contact_name = "test-caps-ft2@" + get_host_name()
listener, port = setup_stream_listener(q, contact_name)
announcer = AvahiAnnouncer(contact_name, "_presence._tcp", port,
txt_record)
# this is the first presence, Salut connects to the contact
e = q.expect('incoming-connection', listener = listener)
incoming = e.connection
# Salut looks up our capabilities
event = q.expect('stream-iq', connection = incoming,
query_ns='http://jabber.org/protocol/disco#info')
query_node = xpath.queryForNodes('/iq/query', event.stanza)[0]
assert query_node.attributes['node'] == \
client + '#' + ver, (query_node.attributes['node'], client, ver)
contact_handle = conn_contacts_iface.GetContactByID(contact_name, [])[0]
# send good reply
result = make_result_iq(event.stanza)
query = result.firstChildElement()
query['node'] = client + '#' + ver
feature = query.addElement('feature')
feature['var'] = "http://telepathy.freedesktop.org/xmpp/pony"
incoming.send(result)
# the FT capability is not announced
e = q.expect('dbus-signal', signal='ContactCapabilitiesChanged')
caps = e.args[0][contact_handle]
assertDoesNotContain(ft_caps, caps)
# check the Contacts interface give the same caps
caps_via_contacts_iface = conn_contacts_iface.GetContactAttributes(
[contact_handle], [cs.CONN_IFACE_CONTACT_CAPS]) \
[contact_handle][cs.CONN_IFACE_CONTACT_CAPS + '/capabilities']
assert caps_via_contacts_iface == caps, caps_via_contacts_iface
# no capabilites announced (assume FT is supported to insure interop)
txt_record = { "txtvers": "1", "status": "avail"}
contact_name = "test-caps-ft-no-capa2@" + get_host_name()
contact_handle = conn_contacts_iface.GetContactByID(contact_name, [])[0]
listener, port = setup_stream_listener(q, contact_name)
announcer = AvahiAnnouncer(contact_name, "_presence._tcp", port,
txt_record)
# FT capa is announced
e = q.expect('dbus-signal', signal='ContactCapabilitiesChanged',
predicate=lambda e: contact_handle in e.args[0].keys())
caps = e.args[0][contact_handle]
assertContains(ft_caps, caps)
# check the Contacts interface give the same caps
caps_via_contacts_iface = conn_contacts_iface.GetContactAttributes(
[contact_handle], [cs.CONN_IFACE_CONTACT_CAPS]) \
[contact_handle][cs.CONN_IFACE_CONTACT_CAPS + '/capabilities']
assert caps_via_contacts_iface == caps, caps_via_contacts_iface
def test(q, bus, conn):
# last value of the "ver" key we resolved. We use it to be sure that the
# modified caps has already be announced.
old_ver = None
conn.Connect()
q.expect('dbus-signal', signal='StatusChanged', args=[0, 0])
# check our own capabilities
self_handle = conn.Properties.Get(cs.CONN, "SelfHandle")
conn_contacts_iface = dbus.Interface(conn, cs.CONN_IFACE_CONTACTS)
caps = conn_contacts_iface.GetContactAttributes(
[self_handle], [cs.CONN_IFACE_CONTACT_CAPS]) \
[self_handle][cs.CONN_IFACE_CONTACT_CAPS + '/capabilities']
assertContains(ft_metadata_caps, caps)
client = 'http://telepathy.freedesktop.org/fake-client'
test_ft_caps_from_contact(q, bus, conn, client)
conn.Disconnect()
q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
if __name__ == '__main__':
exec_test(test)
|
