##// END OF EJS Templates
basic tests for data_pub
MinRK -
Show More
@@ -1,609 +1,630 b''
1 1 # -*- coding: utf-8 -*-
2 2 """test View objects
3 3
4 4 Authors:
5 5
6 6 * Min RK
7 7 """
8 8 #-------------------------------------------------------------------------------
9 9 # Copyright (C) 2011 The IPython Development Team
10 10 #
11 11 # Distributed under the terms of the BSD License. The full license is in
12 12 # the file COPYING, distributed as part of this software.
13 13 #-------------------------------------------------------------------------------
14 14
15 15 #-------------------------------------------------------------------------------
16 16 # Imports
17 17 #-------------------------------------------------------------------------------
18 18
19 19 import sys
20 20 import platform
21 21 import time
22 22 from tempfile import mktemp
23 23 from StringIO import StringIO
24 24
25 25 import zmq
26 26 from nose import SkipTest
27 27
28 28 from IPython.testing import decorators as dec
29 29 from IPython.testing.ipunittest import ParametricTestCase
30 30
31 31 from IPython import parallel as pmod
32 32 from IPython.parallel import error
33 33 from IPython.parallel import AsyncResult, AsyncHubResult, AsyncMapResult
34 34 from IPython.parallel import DirectView
35 35 from IPython.parallel.util import interactive
36 36
37 37 from IPython.parallel.tests import add_engines
38 38
39 39 from .clienttest import ClusterTestCase, crash, wait, skip_without
40 40
41 41 def setup():
42 42 add_engines(3, total=True)
43 43
44 44 class TestView(ClusterTestCase, ParametricTestCase):
45 45
46 46 def setUp(self):
47 47 # On Win XP, wait for resource cleanup, else parallel test group fails
48 48 if platform.system() == "Windows" and platform.win32_ver()[0] == "XP":
49 49 # 1 sec fails. 1.5 sec seems ok. Using 2 sec for margin of safety
50 50 time.sleep(2)
51 51 super(TestView, self).setUp()
52 52
53 53 def test_z_crash_mux(self):
54 54 """test graceful handling of engine death (direct)"""
55 55 raise SkipTest("crash tests disabled, due to undesirable crash reports")
56 56 # self.add_engines(1)
57 57 eid = self.client.ids[-1]
58 58 ar = self.client[eid].apply_async(crash)
59 59 self.assertRaisesRemote(error.EngineError, ar.get, 10)
60 60 eid = ar.engine_id
61 61 tic = time.time()
62 62 while eid in self.client.ids and time.time()-tic < 5:
63 63 time.sleep(.01)
64 64 self.client.spin()
65 65 self.assertFalse(eid in self.client.ids, "Engine should have died")
66 66
67 67 def test_push_pull(self):
68 68 """test pushing and pulling"""
69 69 data = dict(a=10, b=1.05, c=range(10), d={'e':(1,2),'f':'hi'})
70 70 t = self.client.ids[-1]
71 71 v = self.client[t]
72 72 push = v.push
73 73 pull = v.pull
74 74 v.block=True
75 75 nengines = len(self.client)
76 76 push({'data':data})
77 77 d = pull('data')
78 78 self.assertEqual(d, data)
79 79 self.client[:].push({'data':data})
80 80 d = self.client[:].pull('data', block=True)
81 81 self.assertEqual(d, nengines*[data])
82 82 ar = push({'data':data}, block=False)
83 83 self.assertTrue(isinstance(ar, AsyncResult))
84 84 r = ar.get()
85 85 ar = self.client[:].pull('data', block=False)
86 86 self.assertTrue(isinstance(ar, AsyncResult))
87 87 r = ar.get()
88 88 self.assertEqual(r, nengines*[data])
89 89 self.client[:].push(dict(a=10,b=20))
90 90 r = self.client[:].pull(('a','b'), block=True)
91 91 self.assertEqual(r, nengines*[[10,20]])
92 92
93 93 def test_push_pull_function(self):
94 94 "test pushing and pulling functions"
95 95 def testf(x):
96 96 return 2.0*x
97 97
98 98 t = self.client.ids[-1]
99 99 v = self.client[t]
100 100 v.block=True
101 101 push = v.push
102 102 pull = v.pull
103 103 execute = v.execute
104 104 push({'testf':testf})
105 105 r = pull('testf')
106 106 self.assertEqual(r(1.0), testf(1.0))
107 107 execute('r = testf(10)')
108 108 r = pull('r')
109 109 self.assertEqual(r, testf(10))
110 110 ar = self.client[:].push({'testf':testf}, block=False)
111 111 ar.get()
112 112 ar = self.client[:].pull('testf', block=False)
113 113 rlist = ar.get()
114 114 for r in rlist:
115 115 self.assertEqual(r(1.0), testf(1.0))
116 116 execute("def g(x): return x*x")
117 117 r = pull(('testf','g'))
118 118 self.assertEqual((r[0](10),r[1](10)), (testf(10), 100))
119 119
120 120 def test_push_function_globals(self):
121 121 """test that pushed functions have access to globals"""
122 122 @interactive
123 123 def geta():
124 124 return a
125 125 # self.add_engines(1)
126 126 v = self.client[-1]
127 127 v.block=True
128 128 v['f'] = geta
129 129 self.assertRaisesRemote(NameError, v.execute, 'b=f()')
130 130 v.execute('a=5')
131 131 v.execute('b=f()')
132 132 self.assertEqual(v['b'], 5)
133 133
134 134 def test_push_function_defaults(self):
135 135 """test that pushed functions preserve default args"""
136 136 def echo(a=10):
137 137 return a
138 138 v = self.client[-1]
139 139 v.block=True
140 140 v['f'] = echo
141 141 v.execute('b=f()')
142 142 self.assertEqual(v['b'], 10)
143 143
144 144 def test_get_result(self):
145 145 """test getting results from the Hub."""
146 146 c = pmod.Client(profile='iptest')
147 147 # self.add_engines(1)
148 148 t = c.ids[-1]
149 149 v = c[t]
150 150 v2 = self.client[t]
151 151 ar = v.apply_async(wait, 1)
152 152 # give the monitor time to notice the message
153 153 time.sleep(.25)
154 154 ahr = v2.get_result(ar.msg_ids)
155 155 self.assertTrue(isinstance(ahr, AsyncHubResult))
156 156 self.assertEqual(ahr.get(), ar.get())
157 157 ar2 = v2.get_result(ar.msg_ids)
158 158 self.assertFalse(isinstance(ar2, AsyncHubResult))
159 159 c.spin()
160 160 c.close()
161 161
162 162 def test_run_newline(self):
163 163 """test that run appends newline to files"""
164 164 tmpfile = mktemp()
165 165 with open(tmpfile, 'w') as f:
166 166 f.write("""def g():
167 167 return 5
168 168 """)
169 169 v = self.client[-1]
170 170 v.run(tmpfile, block=True)
171 171 self.assertEqual(v.apply_sync(lambda f: f(), pmod.Reference('g')), 5)
172 172
173 173 def test_apply_tracked(self):
174 174 """test tracking for apply"""
175 175 # self.add_engines(1)
176 176 t = self.client.ids[-1]
177 177 v = self.client[t]
178 178 v.block=False
179 179 def echo(n=1024*1024, **kwargs):
180 180 with v.temp_flags(**kwargs):
181 181 return v.apply(lambda x: x, 'x'*n)
182 182 ar = echo(1, track=False)
183 183 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
184 184 self.assertTrue(ar.sent)
185 185 ar = echo(track=True)
186 186 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
187 187 self.assertEqual(ar.sent, ar._tracker.done)
188 188 ar._tracker.wait()
189 189 self.assertTrue(ar.sent)
190 190
191 191 def test_push_tracked(self):
192 192 t = self.client.ids[-1]
193 193 ns = dict(x='x'*1024*1024)
194 194 v = self.client[t]
195 195 ar = v.push(ns, block=False, track=False)
196 196 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
197 197 self.assertTrue(ar.sent)
198 198
199 199 ar = v.push(ns, block=False, track=True)
200 200 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
201 201 ar._tracker.wait()
202 202 self.assertEqual(ar.sent, ar._tracker.done)
203 203 self.assertTrue(ar.sent)
204 204 ar.get()
205 205
206 206 def test_scatter_tracked(self):
207 207 t = self.client.ids
208 208 x='x'*1024*1024
209 209 ar = self.client[t].scatter('x', x, block=False, track=False)
210 210 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
211 211 self.assertTrue(ar.sent)
212 212
213 213 ar = self.client[t].scatter('x', x, block=False, track=True)
214 214 self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
215 215 self.assertEqual(ar.sent, ar._tracker.done)
216 216 ar._tracker.wait()
217 217 self.assertTrue(ar.sent)
218 218 ar.get()
219 219
220 220 def test_remote_reference(self):
221 221 v = self.client[-1]
222 222 v['a'] = 123
223 223 ra = pmod.Reference('a')
224 224 b = v.apply_sync(lambda x: x, ra)
225 225 self.assertEqual(b, 123)
226 226
227 227
228 228 def test_scatter_gather(self):
229 229 view = self.client[:]
230 230 seq1 = range(16)
231 231 view.scatter('a', seq1)
232 232 seq2 = view.gather('a', block=True)
233 233 self.assertEqual(seq2, seq1)
234 234 self.assertRaisesRemote(NameError, view.gather, 'asdf', block=True)
235 235
236 236 @skip_without('numpy')
237 237 def test_scatter_gather_numpy(self):
238 238 import numpy
239 239 from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
240 240 view = self.client[:]
241 241 a = numpy.arange(64)
242 242 view.scatter('a', a, block=True)
243 243 b = view.gather('a', block=True)
244 244 assert_array_equal(b, a)
245 245
246 246 def test_scatter_gather_lazy(self):
247 247 """scatter/gather with targets='all'"""
248 248 view = self.client.direct_view(targets='all')
249 249 x = range(64)
250 250 view.scatter('x', x)
251 251 gathered = view.gather('x', block=True)
252 252 self.assertEqual(gathered, x)
253 253
254 254
255 255 @dec.known_failure_py3
256 256 @skip_without('numpy')
257 257 def test_push_numpy_nocopy(self):
258 258 import numpy
259 259 view = self.client[:]
260 260 a = numpy.arange(64)
261 261 view['A'] = a
262 262 @interactive
263 263 def check_writeable(x):
264 264 return x.flags.writeable
265 265
266 266 for flag in view.apply_sync(check_writeable, pmod.Reference('A')):
267 267 self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
268 268
269 269 view.push(dict(B=a))
270 270 for flag in view.apply_sync(check_writeable, pmod.Reference('B')):
271 271 self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
272 272
273 273 @skip_without('numpy')
274 274 def test_apply_numpy(self):
275 275 """view.apply(f, ndarray)"""
276 276 import numpy
277 277 from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
278 278
279 279 A = numpy.random.random((100,100))
280 280 view = self.client[-1]
281 281 for dt in [ 'int32', 'uint8', 'float32', 'float64' ]:
282 282 B = A.astype(dt)
283 283 C = view.apply_sync(lambda x:x, B)
284 284 assert_array_equal(B,C)
285 285
286 286 @skip_without('numpy')
287 287 def test_push_pull_recarray(self):
288 288 """push/pull recarrays"""
289 289 import numpy
290 290 from numpy.testing.utils import assert_array_equal
291 291
292 292 view = self.client[-1]
293 293
294 294 R = numpy.array([
295 295 (1, 'hi', 0.),
296 296 (2**30, 'there', 2.5),
297 297 (-99999, 'world', -12345.6789),
298 298 ], [('n', int), ('s', '|S10'), ('f', float)])
299 299
300 300 view['RR'] = R
301 301 R2 = view['RR']
302 302
303 303 r_dtype, r_shape = view.apply_sync(interactive(lambda : (RR.dtype, RR.shape)))
304 304 self.assertEqual(r_dtype, R.dtype)
305 305 self.assertEqual(r_shape, R.shape)
306 306 self.assertEqual(R2.dtype, R.dtype)
307 307 self.assertEqual(R2.shape, R.shape)
308 308 assert_array_equal(R2, R)
309 309
310 310 def test_map(self):
311 311 view = self.client[:]
312 312 def f(x):
313 313 return x**2
314 314 data = range(16)
315 315 r = view.map_sync(f, data)
316 316 self.assertEqual(r, map(f, data))
317 317
318 318 def test_map_iterable(self):
319 319 """test map on iterables (direct)"""
320 320 view = self.client[:]
321 321 # 101 is prime, so it won't be evenly distributed
322 322 arr = range(101)
323 323 # ensure it will be an iterator, even in Python 3
324 324 it = iter(arr)
325 325 r = view.map_sync(lambda x:x, arr)
326 326 self.assertEqual(r, list(arr))
327 327
328 328 def test_scatter_gather_nonblocking(self):
329 329 data = range(16)
330 330 view = self.client[:]
331 331 view.scatter('a', data, block=False)
332 332 ar = view.gather('a', block=False)
333 333 self.assertEqual(ar.get(), data)
334 334
335 335 @skip_without('numpy')
336 336 def test_scatter_gather_numpy_nonblocking(self):
337 337 import numpy
338 338 from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
339 339 a = numpy.arange(64)
340 340 view = self.client[:]
341 341 ar = view.scatter('a', a, block=False)
342 342 self.assertTrue(isinstance(ar, AsyncResult))
343 343 amr = view.gather('a', block=False)
344 344 self.assertTrue(isinstance(amr, AsyncMapResult))
345 345 assert_array_equal(amr.get(), a)
346 346
347 347 def test_execute(self):
348 348 view = self.client[:]
349 349 # self.client.debug=True
350 350 execute = view.execute
351 351 ar = execute('c=30', block=False)
352 352 self.assertTrue(isinstance(ar, AsyncResult))
353 353 ar = execute('d=[0,1,2]', block=False)
354 354 self.client.wait(ar, 1)
355 355 self.assertEqual(len(ar.get()), len(self.client))
356 356 for c in view['c']:
357 357 self.assertEqual(c, 30)
358 358
359 359 def test_abort(self):
360 360 view = self.client[-1]
361 361 ar = view.execute('import time; time.sleep(1)', block=False)
362 362 ar2 = view.apply_async(lambda : 2)
363 363 ar3 = view.apply_async(lambda : 3)
364 364 view.abort(ar2)
365 365 view.abort(ar3.msg_ids)
366 366 self.assertRaises(error.TaskAborted, ar2.get)
367 367 self.assertRaises(error.TaskAborted, ar3.get)
368 368
369 369 def test_abort_all(self):
370 370 """view.abort() aborts all outstanding tasks"""
371 371 view = self.client[-1]
372 372 ars = [ view.apply_async(time.sleep, 0.25) for i in range(10) ]
373 373 view.abort()
374 374 view.wait(timeout=5)
375 375 for ar in ars[5:]:
376 376 self.assertRaises(error.TaskAborted, ar.get)
377 377
378 378 def test_temp_flags(self):
379 379 view = self.client[-1]
380 380 view.block=True
381 381 with view.temp_flags(block=False):
382 382 self.assertFalse(view.block)
383 383 self.assertTrue(view.block)
384 384
385 385 @dec.known_failure_py3
386 386 def test_importer(self):
387 387 view = self.client[-1]
388 388 view.clear(block=True)
389 389 with view.importer:
390 390 import re
391 391
392 392 @interactive
393 393 def findall(pat, s):
394 394 # this globals() step isn't necessary in real code
395 395 # only to prevent a closure in the test
396 396 re = globals()['re']
397 397 return re.findall(pat, s)
398 398
399 399 self.assertEqual(view.apply_sync(findall, '\w+', 'hello world'), 'hello world'.split())
400 400
401 401 def test_unicode_execute(self):
402 402 """test executing unicode strings"""
403 403 v = self.client[-1]
404 404 v.block=True
405 405 if sys.version_info[0] >= 3:
406 406 code="a='é'"
407 407 else:
408 408 code=u"a=u'é'"
409 409 v.execute(code)
410 410 self.assertEqual(v['a'], u'é')
411 411
412 412 def test_unicode_apply_result(self):
413 413 """test unicode apply results"""
414 414 v = self.client[-1]
415 415 r = v.apply_sync(lambda : u'é')
416 416 self.assertEqual(r, u'é')
417 417
418 418 def test_unicode_apply_arg(self):
419 419 """test passing unicode arguments to apply"""
420 420 v = self.client[-1]
421 421
422 422 @interactive
423 423 def check_unicode(a, check):
424 424 assert isinstance(a, unicode), "%r is not unicode"%a
425 425 assert isinstance(check, bytes), "%r is not bytes"%check
426 426 assert a.encode('utf8') == check, "%s != %s"%(a,check)
427 427
428 428 for s in [ u'é', u'ßø®∫',u'asdf' ]:
429 429 try:
430 430 v.apply_sync(check_unicode, s, s.encode('utf8'))
431 431 except error.RemoteError as e:
432 432 if e.ename == 'AssertionError':
433 433 self.fail(e.evalue)
434 434 else:
435 435 raise e
436 436
437 437 def test_map_reference(self):
438 438 """view.map(<Reference>, *seqs) should work"""
439 439 v = self.client[:]
440 440 v.scatter('n', self.client.ids, flatten=True)
441 441 v.execute("f = lambda x,y: x*y")
442 442 rf = pmod.Reference('f')
443 443 nlist = list(range(10))
444 444 mlist = nlist[::-1]
445 445 expected = [ m*n for m,n in zip(mlist, nlist) ]
446 446 result = v.map_sync(rf, mlist, nlist)
447 447 self.assertEqual(result, expected)
448 448
449 449 def test_apply_reference(self):
450 450 """view.apply(<Reference>, *args) should work"""
451 451 v = self.client[:]
452 452 v.scatter('n', self.client.ids, flatten=True)
453 453 v.execute("f = lambda x: n*x")
454 454 rf = pmod.Reference('f')
455 455 result = v.apply_sync(rf, 5)
456 456 expected = [ 5*id for id in self.client.ids ]
457 457 self.assertEqual(result, expected)
458 458
459 459 def test_eval_reference(self):
460 460 v = self.client[self.client.ids[0]]
461 461 v['g'] = range(5)
462 462 rg = pmod.Reference('g[0]')
463 463 echo = lambda x:x
464 464 self.assertEqual(v.apply_sync(echo, rg), 0)
465 465
466 466 def test_reference_nameerror(self):
467 467 v = self.client[self.client.ids[0]]
468 468 r = pmod.Reference('elvis_has_left')
469 469 echo = lambda x:x
470 470 self.assertRaisesRemote(NameError, v.apply_sync, echo, r)
471 471
472 472 def test_single_engine_map(self):
473 473 e0 = self.client[self.client.ids[0]]
474 474 r = range(5)
475 475 check = [ -1*i for i in r ]
476 476 result = e0.map_sync(lambda x: -1*x, r)
477 477 self.assertEqual(result, check)
478 478
479 479 def test_len(self):
480 480 """len(view) makes sense"""
481 481 e0 = self.client[self.client.ids[0]]
482 482 yield self.assertEqual(len(e0), 1)
483 483 v = self.client[:]
484 484 yield self.assertEqual(len(v), len(self.client.ids))
485 485 v = self.client.direct_view('all')
486 486 yield self.assertEqual(len(v), len(self.client.ids))
487 487 v = self.client[:2]
488 488 yield self.assertEqual(len(v), 2)
489 489 v = self.client[:1]
490 490 yield self.assertEqual(len(v), 1)
491 491 v = self.client.load_balanced_view()
492 492 yield self.assertEqual(len(v), len(self.client.ids))
493 493 # parametric tests seem to require manual closing?
494 494 self.client.close()
495 495
496 496
497 497 # begin execute tests
498 498
499 499 def test_execute_reply(self):
500 500 e0 = self.client[self.client.ids[0]]
501 501 e0.block = True
502 502 ar = e0.execute("5", silent=False)
503 503 er = ar.get()
504 504 self.assertEqual(str(er), "<ExecuteReply[%i]: 5>" % er.execution_count)
505 505 self.assertEqual(er.pyout['data']['text/plain'], '5')
506 506
507 507 def test_execute_reply_stdout(self):
508 508 e0 = self.client[self.client.ids[0]]
509 509 e0.block = True
510 510 ar = e0.execute("print (5)", silent=False)
511 511 er = ar.get()
512 512 self.assertEqual(er.stdout.strip(), '5')
513 513
514 514 def test_execute_pyout(self):
515 515 """execute triggers pyout with silent=False"""
516 516 view = self.client[:]
517 517 ar = view.execute("5", silent=False, block=True)
518 518
519 519 expected = [{'text/plain' : '5'}] * len(view)
520 520 mimes = [ out['data'] for out in ar.pyout ]
521 521 self.assertEqual(mimes, expected)
522 522
523 523 def test_execute_silent(self):
524 524 """execute does not trigger pyout with silent=True"""
525 525 view = self.client[:]
526 526 ar = view.execute("5", block=True)
527 527 expected = [None] * len(view)
528 528 self.assertEqual(ar.pyout, expected)
529 529
530 530 def test_execute_magic(self):
531 531 """execute accepts IPython commands"""
532 532 view = self.client[:]
533 533 view.execute("a = 5")
534 534 ar = view.execute("%whos", block=True)
535 535 # this will raise, if that failed
536 536 ar.get(5)
537 537 for stdout in ar.stdout:
538 538 lines = stdout.splitlines()
539 539 self.assertEqual(lines[0].split(), ['Variable', 'Type', 'Data/Info'])
540 540 found = False
541 541 for line in lines[2:]:
542 542 split = line.split()
543 543 if split == ['a', 'int', '5']:
544 544 found = True
545 545 break
546 546 self.assertTrue(found, "whos output wrong: %s" % stdout)
547 547
548 548 def test_execute_displaypub(self):
549 549 """execute tracks display_pub output"""
550 550 view = self.client[:]
551 551 view.execute("from IPython.core.display import *")
552 552 ar = view.execute("[ display(i) for i in range(5) ]", block=True)
553 553
554 554 expected = [ {u'text/plain' : unicode(j)} for j in range(5) ]
555 555 for outputs in ar.outputs:
556 556 mimes = [ out['data'] for out in outputs ]
557 557 self.assertEqual(mimes, expected)
558 558
559 559 def test_apply_displaypub(self):
560 560 """apply tracks display_pub output"""
561 561 view = self.client[:]
562 562 view.execute("from IPython.core.display import *")
563 563
564 564 @interactive
565 565 def publish():
566 566 [ display(i) for i in range(5) ]
567 567
568 568 ar = view.apply_async(publish)
569 569 ar.get(5)
570 570 expected = [ {u'text/plain' : unicode(j)} for j in range(5) ]
571 571 for outputs in ar.outputs:
572 572 mimes = [ out['data'] for out in outputs ]
573 573 self.assertEqual(mimes, expected)
574 574
575 575 def test_execute_raises(self):
576 576 """exceptions in execute requests raise appropriately"""
577 577 view = self.client[-1]
578 578 ar = view.execute("1/0")
579 579 self.assertRaisesRemote(ZeroDivisionError, ar.get, 2)
580 580
581 581 @dec.skipif_not_matplotlib
582 582 def test_magic_pylab(self):
583 583 """%pylab works on engines"""
584 584 view = self.client[-1]
585 585 ar = view.execute("%pylab inline")
586 586 # at least check if this raised:
587 587 reply = ar.get(5)
588 588 # include imports, in case user config
589 589 ar = view.execute("plot(rand(100))", silent=False)
590 590 reply = ar.get(5)
591 591 self.assertEqual(len(reply.outputs), 1)
592 592 output = reply.outputs[0]
593 593 self.assertTrue("data" in output)
594 594 data = output['data']
595 595 self.assertTrue("image/png" in data)
596 596
597 597 def test_func_default_func(self):
598 598 """interactively defined function as apply func default"""
599 599 def foo():
600 600 return 'foo'
601 601
602 602 def bar(f=foo):
603 603 return f()
604 604
605 605 view = self.client[-1]
606 606 ar = view.apply_async(bar)
607 607 r = ar.get(10)
608 608 self.assertEquals(r, 'foo')
609 def test_data_pub_single(self):
610 view = self.client[-1]
611 ar = view.execute('\n'.join([
612 'from IPython.zmq.datapub import publish_data',
613 'for i in range(5):',
614 ' publish_data(dict(i=i))'
615 ]), block=False)
616 self.assertTrue(isinstance(ar.data, dict))
617 ar.get(5)
618 self.assertEqual(ar.data, dict(i=4))
619
620 def test_data_pub(self):
621 view = self.client[:]
622 ar = view.execute('\n'.join([
623 'from IPython.zmq.datapub import publish_data',
624 'for i in range(5):',
625 ' publish_data(dict(i=i))'
626 ]), block=False)
627 self.assertTrue(all(isinstance(d, dict) for d in ar.data))
628 ar.get(5)
629 self.assertEqual(ar.data, [dict(i=4)] * len(ar))
609 630
General Comments 0
You need to be logged in to leave comments. Login now