1 | # Copyright (c) Ralph Meijer. |
---|
2 | # See LICENSE for details. |
---|
3 | |
---|
4 | """ |
---|
5 | Tests for L{wokkel.generic}. |
---|
6 | """ |
---|
7 | |
---|
8 | from __future__ import division, absolute_import |
---|
9 | |
---|
10 | import re |
---|
11 | |
---|
12 | from zope.interface.verify import verifyObject |
---|
13 | |
---|
14 | from twisted.python import deprecate |
---|
15 | from twisted.python.compat import unicode |
---|
16 | from twisted.trial import unittest |
---|
17 | from twisted.trial.util import suppress as SUPPRESS |
---|
18 | from twisted.words.xish import domish |
---|
19 | from twisted.words.protocols.jabber.jid import JID |
---|
20 | |
---|
21 | from incremental import Version |
---|
22 | |
---|
23 | from wokkel import generic |
---|
24 | from wokkel.iwokkel import IDisco |
---|
25 | from wokkel.test.helpers import XmlStreamStub |
---|
26 | |
---|
27 | NS_VERSION = 'jabber:iq:version' |
---|
28 | |
---|
29 | class VersionHandlerTest(unittest.TestCase): |
---|
30 | """ |
---|
31 | Tests for L{wokkel.generic.VersionHandler}. |
---|
32 | """ |
---|
33 | |
---|
34 | def setUp(self): |
---|
35 | self.protocol = generic.VersionHandler('Test', '0.1.0') |
---|
36 | |
---|
37 | |
---|
38 | def test_interface(self): |
---|
39 | """ |
---|
40 | L{generic.VersionHandler} implements {IDisco}. |
---|
41 | """ |
---|
42 | verifyObject(IDisco, self.protocol) |
---|
43 | |
---|
44 | |
---|
45 | def test_onVersion(self): |
---|
46 | """ |
---|
47 | Test response to incoming version request. |
---|
48 | """ |
---|
49 | self.stub = XmlStreamStub() |
---|
50 | self.protocol.xmlstream = self.stub.xmlstream |
---|
51 | self.protocol.send = self.stub.xmlstream.send |
---|
52 | self.protocol.connectionInitialized() |
---|
53 | |
---|
54 | iq = domish.Element((None, 'iq')) |
---|
55 | iq['from'] = 'user@example.org/Home' |
---|
56 | iq['to'] = 'example.org' |
---|
57 | iq['type'] = 'get' |
---|
58 | iq.addElement((NS_VERSION, 'query')) |
---|
59 | self.stub.send(iq) |
---|
60 | |
---|
61 | response = self.stub.output[-1] |
---|
62 | self.assertEquals('user@example.org/Home', response['to']) |
---|
63 | self.assertEquals('example.org', response['from']) |
---|
64 | self.assertEquals('result', response['type']) |
---|
65 | self.assertEquals(NS_VERSION, response.query.uri) |
---|
66 | elements = list(domish.generateElementsQNamed(response.query.children, |
---|
67 | 'name', NS_VERSION)) |
---|
68 | self.assertEquals(1, len(elements)) |
---|
69 | self.assertEquals('Test', unicode(elements[0])) |
---|
70 | elements = list(domish.generateElementsQNamed(response.query.children, |
---|
71 | 'version', NS_VERSION)) |
---|
72 | self.assertEquals(1, len(elements)) |
---|
73 | self.assertEquals('0.1.0', unicode(elements[0])) |
---|
74 | |
---|
75 | |
---|
76 | |
---|
77 | class XmlPipeTest(unittest.TestCase): |
---|
78 | """ |
---|
79 | Tests for L{wokkel.generic.XmlPipe}. |
---|
80 | """ |
---|
81 | |
---|
82 | def setUp(self): |
---|
83 | self.pipe = generic.XmlPipe() |
---|
84 | |
---|
85 | |
---|
86 | def test_sendFromSource(self): |
---|
87 | """ |
---|
88 | Send an element from the source and observe it from the sink. |
---|
89 | """ |
---|
90 | def cb(obj): |
---|
91 | called.append(obj) |
---|
92 | |
---|
93 | called = [] |
---|
94 | self.pipe.sink.addObserver('/test[@xmlns="testns"]', cb) |
---|
95 | element = domish.Element(('testns', 'test')) |
---|
96 | self.pipe.source.send(element) |
---|
97 | self.assertEquals([element], called) |
---|
98 | |
---|
99 | |
---|
100 | def test_sendFromSink(self): |
---|
101 | """ |
---|
102 | Send an element from the sink and observe it from the source. |
---|
103 | """ |
---|
104 | def cb(obj): |
---|
105 | called.append(obj) |
---|
106 | |
---|
107 | called = [] |
---|
108 | self.pipe.source.addObserver('/test[@xmlns="testns"]', cb) |
---|
109 | element = domish.Element(('testns', 'test')) |
---|
110 | self.pipe.sink.send(element) |
---|
111 | self.assertEquals([element], called) |
---|
112 | |
---|
113 | |
---|
114 | |
---|
115 | class StanzaTest(unittest.TestCase): |
---|
116 | """ |
---|
117 | Tests for L{generic.Stanza}. |
---|
118 | """ |
---|
119 | |
---|
120 | def test_fromElement(self): |
---|
121 | xml = """ |
---|
122 | <message type='chat' from='other@example.org' to='user@example.org'/> |
---|
123 | """ |
---|
124 | |
---|
125 | stanza = generic.Stanza.fromElement(generic.parseXml(xml)) |
---|
126 | self.assertEqual('chat', stanza.stanzaType) |
---|
127 | self.assertEqual(JID('other@example.org'), stanza.sender) |
---|
128 | self.assertEqual(JID('user@example.org'), stanza.recipient) |
---|
129 | |
---|
130 | |
---|
131 | def test_fromElementChildParser(self): |
---|
132 | """ |
---|
133 | Child elements for which no parser is defined are ignored. |
---|
134 | """ |
---|
135 | xml = """ |
---|
136 | <message from='other@example.org' to='user@example.org'> |
---|
137 | <x xmlns='http://example.org/'/> |
---|
138 | </message> |
---|
139 | """ |
---|
140 | |
---|
141 | class Message(generic.Stanza): |
---|
142 | childParsers = {('http://example.org/', 'x'): '_childParser_x'} |
---|
143 | elements = [] |
---|
144 | |
---|
145 | def _childParser_x(self, element): |
---|
146 | self.elements.append(element) |
---|
147 | |
---|
148 | message = Message.fromElement(generic.parseXml(xml)) |
---|
149 | self.assertEqual(1, len(message.elements)) |
---|
150 | |
---|
151 | |
---|
152 | def test_fromElementChildParserAll(self): |
---|
153 | """ |
---|
154 | Child elements for which no parser is defined are ignored. |
---|
155 | """ |
---|
156 | xml = """ |
---|
157 | <message from='other@example.org' to='user@example.org'> |
---|
158 | <x xmlns='http://example.org/'/> |
---|
159 | </message> |
---|
160 | """ |
---|
161 | |
---|
162 | class Message(generic.Stanza): |
---|
163 | childParsers = {None: '_childParser'} |
---|
164 | elements = [] |
---|
165 | |
---|
166 | def _childParser(self, element): |
---|
167 | self.elements.append(element) |
---|
168 | |
---|
169 | message = Message.fromElement(generic.parseXml(xml)) |
---|
170 | self.assertEqual(1, len(message.elements)) |
---|
171 | |
---|
172 | |
---|
173 | def test_fromElementChildParserUnknown(self): |
---|
174 | """ |
---|
175 | Child elements for which no parser is defined are ignored. |
---|
176 | """ |
---|
177 | xml = """ |
---|
178 | <message from='other@example.org' to='user@example.org'> |
---|
179 | <x xmlns='http://example.org/'/> |
---|
180 | </message> |
---|
181 | """ |
---|
182 | generic.Stanza.fromElement(generic.parseXml(xml)) |
---|
183 | |
---|
184 | |
---|
185 | |
---|
186 | |
---|
187 | class RequestTest(unittest.TestCase): |
---|
188 | """ |
---|
189 | Tests for L{generic.Request}. |
---|
190 | """ |
---|
191 | |
---|
192 | def setUp(self): |
---|
193 | self.request = generic.Request() |
---|
194 | |
---|
195 | |
---|
196 | def test_requestParser(self): |
---|
197 | """ |
---|
198 | The request's child element is passed to requestParser. |
---|
199 | """ |
---|
200 | xml = """ |
---|
201 | <iq type='get'> |
---|
202 | <query xmlns='jabber:iq:version'/> |
---|
203 | </iq> |
---|
204 | """ |
---|
205 | |
---|
206 | class VersionRequest(generic.Request): |
---|
207 | elements = [] |
---|
208 | |
---|
209 | def parseRequest(self, element): |
---|
210 | self.elements.append((element.uri, element.name)) |
---|
211 | |
---|
212 | request = VersionRequest.fromElement(generic.parseXml(xml)) |
---|
213 | self.assertEqual([(NS_VERSION, 'query')], request.elements) |
---|
214 | |
---|
215 | |
---|
216 | def test_toElementStanzaKind(self): |
---|
217 | """ |
---|
218 | A request is an iq stanza. |
---|
219 | """ |
---|
220 | element = self.request.toElement() |
---|
221 | self.assertIdentical(None, element.uri) |
---|
222 | self.assertEquals('iq', element.name) |
---|
223 | |
---|
224 | |
---|
225 | def test_toElementStanzaType(self): |
---|
226 | """ |
---|
227 | The request has type 'get'. |
---|
228 | """ |
---|
229 | self.assertEquals('get', self.request.stanzaType) |
---|
230 | element = self.request.toElement() |
---|
231 | self.assertEquals('get', element.getAttribute('type')) |
---|
232 | |
---|
233 | |
---|
234 | def test_toElementStanzaTypeSet(self): |
---|
235 | """ |
---|
236 | The request has type 'set'. |
---|
237 | """ |
---|
238 | self.request.stanzaType = 'set' |
---|
239 | element = self.request.toElement() |
---|
240 | self.assertEquals('set', element.getAttribute('type')) |
---|
241 | |
---|
242 | |
---|
243 | def test_toElementStanzaID(self): |
---|
244 | """ |
---|
245 | A request, when rendered, has an identifier. |
---|
246 | """ |
---|
247 | element = self.request.toElement() |
---|
248 | self.assertNotIdentical(None, self.request.stanzaID) |
---|
249 | self.assertEquals(self.request.stanzaID, element.getAttribute('id')) |
---|
250 | |
---|
251 | |
---|
252 | def test_toElementRecipient(self): |
---|
253 | """ |
---|
254 | A request without recipient, has no 'to' attribute. |
---|
255 | """ |
---|
256 | self.request = generic.Request(recipient=JID('other@example.org')) |
---|
257 | self.assertEquals(JID('other@example.org'), self.request.recipient) |
---|
258 | element = self.request.toElement() |
---|
259 | self.assertEquals(u'other@example.org', element.getAttribute('to')) |
---|
260 | |
---|
261 | |
---|
262 | def test_toElementRecipientNone(self): |
---|
263 | """ |
---|
264 | A request without recipient, has no 'to' attribute. |
---|
265 | """ |
---|
266 | element = self.request.toElement() |
---|
267 | self.assertFalse(element.hasAttribute('to')) |
---|
268 | |
---|
269 | |
---|
270 | def test_toElementSender(self): |
---|
271 | """ |
---|
272 | A request with sender, has a 'from' attribute. |
---|
273 | """ |
---|
274 | self.request = generic.Request(sender=JID('user@example.org')) |
---|
275 | self.assertEquals(JID('user@example.org'), self.request.sender) |
---|
276 | element = self.request.toElement() |
---|
277 | self.assertEquals(u'user@example.org', element.getAttribute('from')) |
---|
278 | |
---|
279 | |
---|
280 | def test_toElementSenderNone(self): |
---|
281 | """ |
---|
282 | A request without sender, has no 'from' attribute. |
---|
283 | """ |
---|
284 | element = self.request.toElement() |
---|
285 | self.assertFalse(element.hasAttribute('from')) |
---|
286 | |
---|
287 | |
---|
288 | def test_timeoutDefault(self): |
---|
289 | """ |
---|
290 | The default is no timeout. |
---|
291 | """ |
---|
292 | self.assertIdentical(None, self.request.timeout) |
---|
293 | |
---|
294 | |
---|
295 | def test_stanzaTypeInit(self): |
---|
296 | """ |
---|
297 | If stanzaType is passed in __init__, it overrides the class variable. |
---|
298 | """ |
---|
299 | |
---|
300 | class SetRequest(generic.Request): |
---|
301 | stanzaType = 'set' |
---|
302 | |
---|
303 | request = SetRequest(stanzaType='get') |
---|
304 | self.assertEqual('get', request.stanzaType) |
---|
305 | |
---|
306 | |
---|
307 | def test_stanzaTypeClass(self): |
---|
308 | """ |
---|
309 | If stanzaType is not passed in __init__, the class variable is used. |
---|
310 | """ |
---|
311 | |
---|
312 | class SetRequest(generic.Request): |
---|
313 | stanzaType = 'set' |
---|
314 | |
---|
315 | request = SetRequest() |
---|
316 | self.assertEqual('set', request.stanzaType) |
---|
317 | |
---|
318 | |
---|
319 | |
---|
320 | class PrepareIDNNameTests(unittest.TestCase): |
---|
321 | """ |
---|
322 | Tests for L{wokkel.generic.prepareIDNName}. |
---|
323 | """ |
---|
324 | |
---|
325 | suppress = [SUPPRESS(category=DeprecationWarning, |
---|
326 | message=re.escape( |
---|
327 | deprecate.getDeprecationWarningString( |
---|
328 | generic.prepareIDNName, |
---|
329 | Version("wokkel", 18, 0, 0), |
---|
330 | replacement="unicode.encode('idna')")))] |
---|
331 | |
---|
332 | |
---|
333 | def test_deprecated(self): |
---|
334 | """ |
---|
335 | prepareIDNName is deprecated. |
---|
336 | """ |
---|
337 | self.callDeprecated((Version("wokkel", 18, 0, 0), |
---|
338 | "unicode.encode('idna')"), |
---|
339 | generic.prepareIDNName, ("example.com")) |
---|
340 | test_deprecated.suppress = [] |
---|
341 | |
---|
342 | |
---|
343 | def test_unicode(self): |
---|
344 | """ |
---|
345 | A unicode all-ASCII name is converted to an ASCII byte string. |
---|
346 | """ |
---|
347 | name = u"example.com" |
---|
348 | result = generic.prepareIDNName(name) |
---|
349 | self.assertEqual(b"example.com", result) |
---|
350 | |
---|
351 | |
---|
352 | def test_unicodeNonASCII(self): |
---|
353 | """ |
---|
354 | A unicode with non-ASCII is converted to its ACE equivalent. |
---|
355 | """ |
---|
356 | name = u"\u00e9chec.example.com" |
---|
357 | result = generic.prepareIDNName(name) |
---|
358 | self.assertEqual(b"xn--chec-9oa.example.com", result) |
---|
359 | |
---|
360 | |
---|
361 | def test_unicodeHalfwidthIdeographicFullStop(self): |
---|
362 | """ |
---|
363 | Exotic dots in unicode names are converted to Full Stop. |
---|
364 | """ |
---|
365 | name = u"\u00e9chec.example\uff61com" |
---|
366 | result = generic.prepareIDNName(name) |
---|
367 | self.assertEqual(b"xn--chec-9oa.example.com", result) |
---|
368 | |
---|
369 | |
---|
370 | def test_unicodeTrailingDot(self): |
---|
371 | """ |
---|
372 | Unicode names with trailing dots retain the trailing dot. |
---|
373 | |
---|
374 | L{encodings.idna.ToASCII} doesn't allow the empty string as the input, |
---|
375 | hence the implementation needs to strip a trailing dot, and re-add it |
---|
376 | after encoding the labels. |
---|
377 | """ |
---|
378 | name = u"example.com." |
---|
379 | result = generic.prepareIDNName(name) |
---|
380 | self.assertEqual(b"example.com.", result) |
---|