Mercurial > libervia-pubsub

comparison idavoll/test/test_gateway.py @ 183:c21b986cff30

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Implement HTTP client to gateway and implement functional tests with it.
author Ralph Meijer <ralphm@ik.nu>
date Fri, 11 Apr 2008 14:41:16 +0000
parents
children 9038908dc2f5
comparison
equal deleted inserted replaced
182:4aa29b1a8c67 183:c21b986cff30
1 # Copyright (c) 2003-2008 Ralph Meijer
2 # See LICENSE for details.
3
4 """
5 Tests for L{idavoll.gateway}.
6
7 Note that some tests are functional tests that require a running idavoll
8 service.
9 """
10
11 from twisted.internet import defer
12 from twisted.trial import unittest
13 from twisted.web import error
14 from twisted.words.xish import domish
15
16 from idavoll import gateway
17
18 AGENT = "Idavoll Test Script"
19 NS_ATOM = "http://www.w3.org/2005/Atom"
20
21 entry = domish.Element((NS_ATOM, 'entry'))
22 entry.addElement("id", content="urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a")
23 entry.addElement("title", content="Atom-Powered Robots Run Amok")
24 entry.addElement("author").addElement("name", content="John Doe")
25 entry.addElement("content", content="Some text.")
26
27 #baseURI = "http://pubsub-test.ik.nu/"
28 baseURI = "http://localhost:8086/"
29 componentJID = "test.ik.nu"
30
31 class GatewayTest(unittest.TestCase):
32
33 def setUp(self):
34 self.client = gateway.GatewayClient(baseURI)
35 self.client.startService()
36
37 def tearDown(self):
38 self.client.stopService()
39
40 def test_create(self):
41
42 def cb(response):
43 self.assertIn('uri', response)
44
45 d = self.client.create()
46 d.addCallback(cb)
47 return d
48
49 def test_publish(self):
50
51 def cb(response):
52 self.assertIn('uri', response)
53
54 d = self.client.publish(entry)
55 d.addCallback(cb)
56 return d
57
58 def test_publishExistingNode(self):
59
60 def cb2(response, xmppURI):
61 self.assertEquals(xmppURI, response['uri'])
62
63 def cb1(response):
64 xmppURI = response['uri']
65 d = self.client.publish(entry, xmppURI)
66 d.addCallback(cb2, xmppURI)
67 return d
68
69 d = self.client.create()
70 d.addCallback(cb1)
71 return d
72
73 def test_publishNonExisting(self):
74 def cb(err):
75 self.assertEqual('404', err.status)
76
77 d = self.client.publish(entry, 'xmpp:%s?node=test' % componentJID)
78 self.assertFailure(d, error.Error)
79 d.addCallback(cb)
80 return d
81
82 def test_list(self):
83 d = self.client.listNodes()
84 return d
85
86 def test_subscribe(self):
87 def cb(response):
88 xmppURI = response['uri']
89 d = self.client.subscribe(xmppURI)
90 return d
91
92 d = self.client.create()
93 d.addCallback(cb)
94 return d
95
96 def test_subscribeGetNotification(self):
97
98 def onNotification(data, headers):
99 self.client.deferred.callback(None)
100
101 def cb(response):
102 xmppURI = response['uri']
103 d = self.client.subscribe(xmppURI)
104 d.addCallback(lambda _: xmppURI)
105 return d
106
107 def cb2(xmppURI):
108 d = self.client.publish(entry, xmppURI)
109 return d
110
111
112 self.client.callback = onNotification
113 self.client.deferred = defer.Deferred()
114 d = self.client.create()
115 d.addCallback(cb)
116 d.addCallback(cb2)
117 return defer.gatherResults([d, self.client.deferred])
118
119 def test_subscribeGetDelayedNotification(self):
120
121 def onNotification(data, headers):
122 self.client.deferred.callback(None)
123
124 def cb(response):
125 xmppURI = response['uri']
126 self.assertNot(self.client.deferred.called)
127 d = self.client.publish(entry, xmppURI)
128 d.addCallback(lambda _: xmppURI)
129 return d
130
131 def cb2(xmppURI):
132 d = self.client.subscribe(xmppURI)
133 return d
134
135
136 self.client.callback = onNotification
137 self.client.deferred = defer.Deferred()
138 d = self.client.create()
139 d.addCallback(cb)
140 d.addCallback(cb2)
141 return defer.gatherResults([d, self.client.deferred])
142
143 def test_subscribeNonExisting(self):
144 def cb(err):
145 self.assertEqual('404', err.status)
146
147 d = self.client.subscribe('xmpp:%s?node=test' % componentJID)
148 self.assertFailure(d, error.Error)
149 d.addCallback(cb)
150 return d