test_asyncresult.py
342 lines
| 11.9 KiB
| text/x-python
|
PythonLexer
|
r16739 | """Tests for asyncresult.py""" | ||
|
r4018 | |||
|
r16739 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r3664 | |||
|
r6471 | import time | ||
|
r3664 | |||
|
r8140 | import nose.tools as nt | ||
|
r7324 | from IPython.utils.io import capture_output | ||
|
r3664 | |||
|
r7324 | from IPython.parallel.error import TimeoutError | ||
|
r6471 | from IPython.parallel import error, Client | ||
|
r3666 | from IPython.parallel.tests import add_engines | ||
|
r7324 | from .clienttest import ClusterTestCase | ||
|
r13361 | from IPython.utils.py3compat import iteritems | ||
|
r3664 | |||
def setup(): | ||||
|
r6162 | add_engines(2, total=True) | ||
|
r3664 | |||
def wait(n): | ||||
import time | ||||
time.sleep(n) | ||||
return n | ||||
|
r11362 | def echo(x): | ||
return x | ||||
|
r3664 | class AsyncResultTest(ClusterTestCase): | ||
|
r6426 | def test_single_result_view(self): | ||
"""various one-target views get the right value for single_result""" | ||||
|
r3664 | eid = self.client.ids[-1] | ||
ar = self.client[eid].apply_async(lambda : 42) | ||||
|
r7874 | self.assertEqual(ar.get(), 42) | ||
|
r3664 | ar = self.client[[eid]].apply_async(lambda : 42) | ||
|
r7874 | self.assertEqual(ar.get(), [42]) | ||
|
r3664 | ar = self.client[-1:].apply_async(lambda : 42) | ||
|
r7874 | self.assertEqual(ar.get(), [42]) | ||
|
r3664 | |||
def test_get_after_done(self): | ||||
ar = self.client[-1].apply_async(lambda : 42) | ||||
ar.wait() | ||||
self.assertTrue(ar.ready()) | ||||
|
r7874 | self.assertEqual(ar.get(), 42) | ||
self.assertEqual(ar.get(), 42) | ||||
|
r3664 | |||
def test_get_before_done(self): | ||||
ar = self.client[-1].apply_async(wait, 0.1) | ||||
self.assertRaises(TimeoutError, ar.get, 0) | ||||
ar.wait(0) | ||||
self.assertFalse(ar.ready()) | ||||
|
r7874 | self.assertEqual(ar.get(), 0.1) | ||
|
r3664 | |||
def test_get_after_error(self): | ||||
ar = self.client[-1].apply_async(lambda : 1/0) | ||||
|
r4155 | ar.wait(10) | ||
|
r3664 | self.assertRaisesRemote(ZeroDivisionError, ar.get) | ||
self.assertRaisesRemote(ZeroDivisionError, ar.get) | ||||
self.assertRaisesRemote(ZeroDivisionError, ar.get_dict) | ||||
def test_get_dict(self): | ||||
n = len(self.client) | ||||
ar = self.client[:].apply_async(lambda : 5) | ||||
|
r7874 | self.assertEqual(ar.get(), [5]*n) | ||
|
r3664 | d = ar.get_dict() | ||
|
r7874 | self.assertEqual(sorted(d.keys()), sorted(self.client.ids)) | ||
|
r13361 | for eid,r in iteritems(d): | ||
|
r7874 | self.assertEqual(r, 5) | ||
|
r5222 | |||
|
r11362 | def test_get_dict_single(self): | ||
view = self.client[-1] | ||||
|
r13383 | for v in (list(range(5)), 5, ('abc', 'def'), 'string'): | ||
|
r11362 | ar = view.apply_async(echo, v) | ||
self.assertEqual(ar.get(), v) | ||||
d = ar.get_dict() | ||||
self.assertEqual(d, {view.targets : v}) | ||||
def test_get_dict_bad(self): | ||||
ar = self.client[:].apply_async(lambda : 5) | ||||
ar2 = self.client[:].apply_async(lambda : 5) | ||||
ar = self.client.get_result(ar.msg_ids + ar2.msg_ids) | ||||
self.assertRaises(ValueError, ar.get_dict) | ||||
|
r5222 | def test_list_amr(self): | ||
ar = self.client.load_balanced_view().map_async(wait, [0.1]*5) | ||||
rlist = list(ar) | ||||
def test_getattr(self): | ||||
ar = self.client[:].apply_async(wait, 0.5) | ||||
|
r8105 | self.assertEqual(ar.engine_id, [None] * len(ar)) | ||
|
r5222 | self.assertRaises(AttributeError, lambda : ar._foo) | ||
self.assertRaises(AttributeError, lambda : ar.__length_hint__()) | ||||
self.assertRaises(AttributeError, lambda : ar.foo) | ||||
self.assertFalse(hasattr(ar, '__length_hint__')) | ||||
self.assertFalse(hasattr(ar, 'foo')) | ||||
|
r8105 | self.assertTrue(hasattr(ar, 'engine_id')) | ||
|
r5222 | ar.get(5) | ||
self.assertRaises(AttributeError, lambda : ar._foo) | ||||
self.assertRaises(AttributeError, lambda : ar.__length_hint__()) | ||||
self.assertRaises(AttributeError, lambda : ar.foo) | ||||
self.assertTrue(isinstance(ar.engine_id, list)) | ||||
|
r7874 | self.assertEqual(ar.engine_id, ar['engine_id']) | ||
|
r5222 | self.assertFalse(hasattr(ar, '__length_hint__')) | ||
self.assertFalse(hasattr(ar, 'foo')) | ||||
self.assertTrue(hasattr(ar, 'engine_id')) | ||||
def test_getitem(self): | ||||
ar = self.client[:].apply_async(wait, 0.5) | ||||
|
r8105 | self.assertEqual(ar['engine_id'], [None] * len(ar)) | ||
self.assertRaises(KeyError, lambda : ar['foo']) | ||||
|
r5222 | ar.get(5) | ||
self.assertRaises(KeyError, lambda : ar['foo']) | ||||
self.assertTrue(isinstance(ar['engine_id'], list)) | ||||
|
r7874 | self.assertEqual(ar.engine_id, ar['engine_id']) | ||
|
r5222 | |||
def test_single_result(self): | ||||
ar = self.client[-1].apply_async(wait, 0.5) | ||||
|
r8105 | self.assertRaises(KeyError, lambda : ar['foo']) | ||
self.assertEqual(ar['engine_id'], None) | ||||
|
r5222 | self.assertTrue(ar.get(5) == 0.5) | ||
self.assertTrue(isinstance(ar['engine_id'], int)) | ||||
self.assertTrue(isinstance(ar.engine_id, int)) | ||||
|
r7874 | self.assertEqual(ar.engine_id, ar['engine_id']) | ||
|
r6426 | |||
def test_abort(self): | ||||
e = self.client[-1] | ||||
ar = e.execute('import time; time.sleep(1)', block=False) | ||||
ar2 = e.apply_async(lambda : 2) | ||||
ar2.abort() | ||||
self.assertRaises(error.TaskAborted, ar2.get) | ||||
ar.get() | ||||
|
r6471 | |||
def test_len(self): | ||||
v = self.client.load_balanced_view() | ||||
|
r13383 | ar = v.map_async(lambda x: x, list(range(10))) | ||
|
r7874 | self.assertEqual(len(ar), 10) | ||
|
r13383 | ar = v.apply_async(lambda x: x, list(range(10))) | ||
|
r7874 | self.assertEqual(len(ar), 1) | ||
|
r13383 | ar = self.client[:].apply_async(lambda x: x, list(range(10))) | ||
|
r7874 | self.assertEqual(len(ar), len(self.client.ids)) | ||
|
r6471 | |||
def test_wall_time_single(self): | ||||
v = self.client.load_balanced_view() | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
self.assertRaises(TimeoutError, getattr, ar, 'wall_time') | ||||
ar.get(2) | ||||
self.assertTrue(ar.wall_time < 1.) | ||||
self.assertTrue(ar.wall_time > 0.2) | ||||
def test_wall_time_multi(self): | ||||
self.minimum_engines(4) | ||||
v = self.client[:] | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
self.assertRaises(TimeoutError, getattr, ar, 'wall_time') | ||||
ar.get(2) | ||||
self.assertTrue(ar.wall_time < 1.) | ||||
self.assertTrue(ar.wall_time > 0.2) | ||||
def test_serial_time_single(self): | ||||
v = self.client.load_balanced_view() | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
self.assertRaises(TimeoutError, getattr, ar, 'serial_time') | ||||
ar.get(2) | ||||
|
r6890 | self.assertTrue(ar.serial_time < 1.) | ||
|
r6471 | self.assertTrue(ar.serial_time > 0.2) | ||
def test_serial_time_multi(self): | ||||
self.minimum_engines(4) | ||||
v = self.client[:] | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
self.assertRaises(TimeoutError, getattr, ar, 'serial_time') | ||||
ar.get(2) | ||||
self.assertTrue(ar.serial_time < 2.) | ||||
self.assertTrue(ar.serial_time > 0.8) | ||||
def test_elapsed_single(self): | ||||
v = self.client.load_balanced_view() | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
while not ar.ready(): | ||||
time.sleep(0.01) | ||||
|
r6890 | self.assertTrue(ar.elapsed < 1) | ||
self.assertTrue(ar.elapsed < 1) | ||||
|
r6471 | ar.get(2) | ||
def test_elapsed_multi(self): | ||||
v = self.client[:] | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
while not ar.ready(): | ||||
time.sleep(0.01) | ||||
|
r6890 | self.assertTrue(ar.elapsed < 1) | ||
self.assertTrue(ar.elapsed < 1) | ||||
|
r6471 | ar.get(2) | ||
def test_hubresult_timestamps(self): | ||||
self.minimum_engines(4) | ||||
v = self.client[:] | ||||
ar = v.apply_async(time.sleep, 0.25) | ||||
ar.get(2) | ||||
rc2 = Client(profile='iptest') | ||||
# must have try/finally to close second Client, otherwise | ||||
# will have dangling sockets causing problems | ||||
try: | ||||
time.sleep(0.25) | ||||
hr = rc2.get_result(ar.msg_ids) | ||||
self.assertTrue(hr.elapsed > 0., "got bad elapsed: %s" % hr.elapsed) | ||||
hr.get(1) | ||||
self.assertTrue(hr.wall_time < ar.wall_time + 0.2, "got bad wall_time: %s > %s" % (hr.wall_time, ar.wall_time)) | ||||
|
r7874 | self.assertEqual(hr.serial_time, ar.serial_time) | ||
|
r6471 | finally: | ||
rc2.close() | ||||
|
r5222 | |||
|
r7129 | def test_display_empty_streams_single(self): | ||
"""empty stdout/err are not displayed (single result)""" | ||||
self.minimum_engines(1) | ||||
v = self.client[-1] | ||||
ar = v.execute("print (5555)") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs() | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual('5555\n', io.stdout) | ||||
|
r7129 | |||
ar = v.execute("a=5") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs() | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual(io.stdout, '') | ||||
|
r7129 | |||
def test_display_empty_streams_type(self): | ||||
"""empty stdout/err are not displayed (groupby type)""" | ||||
self.minimum_engines(1) | ||||
v = self.client[:] | ||||
ar = v.execute("print (5555)") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs() | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual(io.stdout.count('5555'), len(v), io.stdout) | ||||
|
r7239 | self.assertFalse('\n\n' in io.stdout, io.stdout) | ||
|
r7874 | self.assertEqual(io.stdout.count('[stdout:'), len(v), io.stdout) | ||
|
r7129 | |||
ar = v.execute("a=5") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs() | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual(io.stdout, '') | ||||
|
r7129 | |||
def test_display_empty_streams_engine(self): | ||||
"""empty stdout/err are not displayed (groupby engine)""" | ||||
self.minimum_engines(1) | ||||
v = self.client[:] | ||||
ar = v.execute("print (5555)") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs('engine') | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual(io.stdout.count('5555'), len(v), io.stdout) | ||||
|
r7239 | self.assertFalse('\n\n' in io.stdout, io.stdout) | ||
|
r7874 | self.assertEqual(io.stdout.count('[stdout:'), len(v), io.stdout) | ||
|
r7129 | |||
ar = v.execute("a=5") | ||||
ar.get(5) | ||||
with capture_output() as io: | ||||
ar.display_outputs('engine') | ||||
|
r7874 | self.assertEqual(io.stderr, '') | ||
self.assertEqual(io.stdout, '') | ||||
|
r8140 | |||
def test_await_data(self): | ||||
"""asking for ar.data flushes outputs""" | ||||
self.minimum_engines(1) | ||||
|
r7129 | |||
|
r8140 | v = self.client[-1] | ||
ar = v.execute('\n'.join([ | ||||
"import time", | ||||
|
r9372 | "from IPython.kernel.zmq.datapub import publish_data", | ||
|
r8140 | "for i in range(5):", | ||
" publish_data(dict(i=i))", | ||||
" time.sleep(0.1)", | ||||
]), block=False) | ||||
found = set() | ||||
tic = time.time() | ||||
# timeout after 10s | ||||
while time.time() <= tic + 10: | ||||
if ar.data: | ||||
|
r13482 | i = ar.data['i'] | ||
found.add(i) | ||||
if i == 4: | ||||
|
r8140 | break | ||
time.sleep(0.05) | ||||
ar.get(5) | ||||
nt.assert_in(4, found) | ||||
self.assertTrue(len(found) > 1, "should have seen data multiple times, but got: %s" % found) | ||||
|
r11415 | |||
def test_not_single_result(self): | ||||
save_build = self.client._build_targets | ||||
def single_engine(*a, **kw): | ||||
idents, targets = save_build(*a, **kw) | ||||
return idents[:1], targets[:1] | ||||
ids = single_engine('all')[1] | ||||
self.client._build_targets = single_engine | ||||
for targets in ('all', None, ids): | ||||
dv = self.client.direct_view(targets=targets) | ||||
ar = dv.apply_async(lambda : 5) | ||||
self.assertEqual(ar.get(10), [5]) | ||||
self.client._build_targets = save_build | ||||
|
r16739 | |||
def test_owner_pop(self): | ||||
self.minimum_engines(1) | ||||
view = self.client[-1] | ||||
ar = view.apply_async(lambda : 1) | ||||
ar.get() | ||||
msg_id = ar.msg_ids[0] | ||||
self.assertNotIn(msg_id, self.client.results) | ||||
self.assertNotIn(msg_id, self.client.metadata) | ||||
def test_non_owner(self): | ||||
self.minimum_engines(1) | ||||
view = self.client[-1] | ||||
ar = view.apply_async(lambda : 1) | ||||
ar.owner = False | ||||
ar.get() | ||||
msg_id = ar.msg_ids[0] | ||||
self.assertIn(msg_id, self.client.results) | ||||
self.assertIn(msg_id, self.client.metadata) | ||||
ar2 = self.client.get_result(msg_id, owner=True) | ||||
self.assertIs(type(ar2), type(ar)) | ||||
self.assertTrue(ar2.owner) | ||||
self.assertEqual(ar.get(), ar2.get()) | ||||
ar2.get() | ||||
self.assertNotIn(msg_id, self.client.results) | ||||
self.assertNotIn(msg_id, self.client.metadata) | ||||
|
r11415 | |||
|
r3664 | |||