mitogen/tests/call_function_test.py

85 lines
2.1 KiB
Python

import unittest
import mitogen.core
import mitogen.master
class CrazyType(object):
pass
def function_that_fails():
raise ValueError('exception text')
def func_with_bad_return_value():
return CrazyType()
def func_returns_dead():
return mitogen.core._DEAD
def func_accepts_returns_context(context):
return context
class CallFunctionTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
super(CallFunctionTest, cls).setUpClass()
cls.broker = mitogen.master.Broker()
cls.router = mitogen.master.Router(cls.broker)
cls.local = cls.router.local()
@classmethod
def tearDownClass(cls):
super(CallFunctionTest, cls).tearDownClass()
cls.broker.shutdown()
cls.broker.join()
def test_crashes(self):
try:
self.local.call(function_that_fails)
assert 0, 'call didnt fail'
except mitogen.core.CallError, e:
pass
s = str(e)
etype, _, s = s.partition(': ')
assert etype == 'exceptions.ValueError'
msg, _, s = s.partition('\n')
assert msg == 'exception text'
# Traceback
assert len(s) > 0
def test_bad_return_value(self):
try:
self.local.call(func_with_bad_return_value)
assert 0, 'call didnt fail'
except mitogen.core.StreamError, e:
pass
assert e[0] == "attempted unpickle from 'call_function_test'"
def test_returns_dead(self):
assert mitogen.core._DEAD == self.local.call(func_returns_dead)
def test_aborted_on_context_disconnect(self):
assert 0, 'todo'
def test_aborted_on_context_hang_deadline(self):
# related: how to treat context after a function call hangs
assert 0, 'todo'
def test_aborted_on_local_broker_shutdown(self):
assert 0, 'todo'
def test_accepts_returns_context(self):
context = self.local.call(func_accepts_returns_context, self.local)
assert context is not self.local
assert context.context_id == self.local.context_id
assert context.name == self.local.name