« back to all changes in this revision
Viewing changes to juju/rapi/transport/tests/test_ws.py
- browse files at revision 624
- compare with another revision
- download diff
- download tarball
- view history from revision 624
- Committer: Kapil Thangavelu
- Date: 2013-01-20 12:08:56 UTC
- mfrom: (616.2.12 rapi-annotation)
- Revision ID: kapil@canonical.com-20130120120856-j3bx91xk7ev4hu7u
Merged rapi-annotation into rapi-rollup.
added
removed
juju/rapi/transport/tests/test_ws.py
1
1
import json
2
2
import logging
3
3
4
5
4
from twisted.internet.defer import inlineCallbacks, Deferred
6
5
7
6
from juju.rapi.transport.ws import WebSocketAPIFactory
8
7
from juju.providers.dummy import MachineProvider
9
8
9
from juju.state.auth import make_ace
10
from juju.state.topology import InternalTopology
11
from juju.state.security import Principal
12
10
13
from juju.rapi.tests.common import ContextTestBase
11
from juju.state.topology import InternalTopology
12
14
from juju.state.tests.common import StateTestBase
13
15
from juju.tests.common import get_test_zookeeper_address
14
16
112
114
@inlineCallbacks
113
115
def test_status(self):
114
116
d = self.transport.watch_kv('op', 'status')
117
115
118
self.ws.connectionMade()
119
self.ws.context.authenticated = True
120
116
121
self.ws.dataReceived(json.dumps(dict(
117
122
op='status')))
118
123
msg = yield d
145
150
yield self.context.deploy("precise/mysql")
146
151
yield self.context.update_annotations(
147
152
"mysql", {'gui.x': 21, 'gui.y': 64})
153
148
154
yield self.stream_manager.pump()
149
155
yield self.stream_manager.pump()
150
156
151
157
d = self.transport.watch_kv('op', 'delta')
152
158
self.ws.connectionMade()
159
self.ws.context.authenticated = True
153
160
yield d
161
154
162
delta = self.transport.stream[-1]
155
163
self.assertEqual(
156
164
delta,
171
179
)
172
180
173
181
@inlineCallbacks
182
def test_delta_after_login(self):
183
"""No delta events are sent till after login.
184
"""
185
# Setup for login and service deploy
186
u, p = "admin", "cekret"
187
self.client.create("/login")
188
yield self.mock_store_charms(["precise/wordpress"])
189
self.mocker.replay()
190
191
self.ws.connectionMade()
192
yield self.context.deploy("precise/wordpress")
193
194
# Even if we pump the delta stream now, the connected client
195
# sees nothing till after login.
196
yield self.stream_manager.pump()
197
198
greeting = self.transport.stream[-1]
199
self.assertEqual(greeting,
200
{u'state': u'login-required',
201
u'version': 0,
202
u'extensions': [],
203
u'ready': True,
204
u'default_series': u'precise',
205
u'provider_type': u'dummy'})
206
207
login_d = self.transport.watch_kv('op', 'login')
208
delta_d = self.transport.watch_kv('op', 'delta')
209
210
self.ws.dataReceived(json.dumps(dict(
211
op='login', user=u, password=p)))
212
213
yield login_d
214
yield delta_d
215
216
delta = self.transport.stream[-1]
217
self.assertEqual(delta['op'], 'delta')
218
219
@inlineCallbacks
174
220
def test_delta_stream(self):
175
221
yield self.mock_store_charms(["precise/wordpress", "precise/mysql"])
176
222
self.mocker.replay()
178
224
# There's always an implicit initial pump of current state made
179
225
# on connection. But no initial state for this test.
180
226
self.ws.connectionMade()
227
self.ws.context.authenticated = True
181
228
182
229
# Modify state and send events out.
183
230
yield self.context.deploy("precise/mysql")
186
233
yield self.stream_manager.pump()
187
234
188
235
delta = self.transport.stream[-1]
236
189
237
self.assertEqual(delta['op'], 'delta')
190
238
self.assertEqual(delta['result'], [
191
239
[u'service', u'add', {u'id': 'wordpress',
Loggerhead is a web-based interface for Breezy
Version: 2.0.1