Rooba
/
mitogen
mirror of https://github.com/mitogen-hq/mitogen.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use specific TestCase assertions throughout tests 

e.g. assert x == y -> self.assertEqual(x, y);
self.assertTrue(isinstance(x, y)) -> self.assertIsInstance(x, y)

These specific methods give more useful errors in the case of a test
failure.
This commit is contained in:
Alex Willmer 2017-11-11 23:08:08 +00:00 committed by David Wilson
parent df9556d1d9
commit e8e023ce59
11 changed files with 50 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
tests/call_function_test.py
View File

@ -39,7 +39,7 @@ class CallFunctionTest(testlib.RouterMixin, testlib.TestCase):
self.local = self.router.local()
def test_succeeds(self):
assert 3 == self.local.call(function_that_adds_numbers, 1, 2)
self.assertEqual(3, self.local.call(function_that_adds_numbers, 1, 2))
def test_crashes(self):
exc = self.assertRaises(mitogen.core.CallError,
@ -47,13 +47,13 @@ class CallFunctionTest(testlib.RouterMixin, testlib.TestCase):
s = str(exc)
etype, _, s = s.partition(': ')
assert etype == 'exceptions.ValueError'
self.assertEqual(etype, 'exceptions.ValueError')
msg, _, s = s.partition('\n')
assert msg == 'exception text'
self.assertEqual(msg, 'exception text')
# Traceback
assert len(s) > 0
self.assertGreater(len(s), 0)
def test_bad_return_value(self):
exc = self.assertRaises(mitogen.core.StreamError,
@ -61,7 +61,7 @@ class CallFunctionTest(testlib.RouterMixin, testlib.TestCase):
self.assertEquals(exc[0], "cannot unpickle '__main__'/'CrazyType'")
def test_returns_dead(self):
assert mitogen.core._DEAD == self.local.call(func_returns_dead)
self.assertEqual(mitogen.core._DEAD, self.local.call(func_returns_dead))
def test_aborted_on_local_context_disconnect(self):
stream = self.router._stream_by_id[self.local.context_id]
@ -83,9 +83,9 @@ class CallFunctionTest(testlib.RouterMixin, testlib.TestCase):
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
self.assertIsNot(context, self.local)
self.assertEqual(context.context_id, self.local.context_id)
self.assertEqual(context.name, self.local.name)
if __name__ == '__main__':

10
tests/channel_test.py
View File

@ -9,11 +9,11 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# issue 32
l1 = self.router.local()
chan = mitogen.core.Channel(self.router, l1, 123)
assert chan.router == self.router
assert chan.context == l1
assert chan.dst_handle == 123
assert chan.handle is not None
assert chan.handle > 0
self.assertEqual(chan.router, self.router)
self.assertEqual(chan.context, l1)
self.assertEqual(chan.dst_handle, 123)
self.assertIsNotNone(chan.handle)
self.assertGreater(chan.handle, 0)
if __name__ == '__main__':

14
tests/fakessh_test.py
View File

@ -21,9 +21,9 @@ class RsyncTest(testlib.DockerMixin, unittest.TestCase):
testlib.data_path('.'), 'target:/tmp/data'
])
assert return_code == 0
assert context.call(os.path.exists, '/tmp/data')
assert context.call(os.path.exists, '/tmp/data/simple_pkg/a.py')
self.assertEqual(return_code, 0)
self.assertTrue(context.call(os.path.exists, '/tmp/data'))
self.assertTrue(context.call(os.path.exists, '/tmp/data/simple_pkg/a.py'))
def test_rsync_between_direct_children(self):
# master -> SSH -> has-sudo-pubkey -> rsync(.ssh) -> master ->
@ -52,9 +52,11 @@ class RsyncTest(testlib.DockerMixin, unittest.TestCase):
'rsync', '--progress', '-vvva', '.ssh/', 'target:' + dest_path
])
assert return_code == 0
assert pubkey_acct.call(os.path.getsize, '.ssh/authorized_keys') == \
webapp_acct.call(os.path.getsize, dest_path + '/authorized_keys')
self.assertEqual(return_code, 0)
self.assertEqual(
pubkey_acct.call(os.path.getsize, '.ssh/authorized_keys'),
webapp_acct.call(os.path.getsize, dest_path + '/authorized_keys'),
)
if __name__ == '__main__':

2
tests/id_allocation_test.py
View File

@ -9,7 +9,7 @@ class SlaveTest(testlib.RouterMixin, testlib.TestCase):
def test_slave_allocates_id(self):
context = self.router.local()
id_ = context.call(id_allocation.allocate_an_id)
assert id_ == (self.router.id_allocator.next_id - 1)
self.assertEqual(id_, self.router.id_allocator.next_id - 1)
if __name__ == '__main__':

12
tests/importer_test.py
View File

@ -39,8 +39,8 @@ class LoadModuleTest(ImporterMixin, testlib.TestCase):
def test_module_added_to_sys_modules(self):
self.context.send_await.return_value = self.response
mod = self.importer.load_module(self.modname)
self.assertTrue(sys.modules[self.modname] is mod)
self.assertTrue(isinstance(mod, types.ModuleType))
self.assertIs(sys.modules[self.modname], mod)
self.assertIsInstance(mod, types.ModuleType)
def test_module_file_set(self):
self.context.send_await.return_value = self.response
@ -50,12 +50,12 @@ class LoadModuleTest(ImporterMixin, testlib.TestCase):
def test_module_loader_set(self):
self.context.send_await.return_value = self.response
mod = self.importer.load_module(self.modname)
self.assertTrue(mod.__loader__ is self.importer)
self.assertIs(mod.__loader__, self.importer)
def test_module_package_unset(self):
self.context.send_await.return_value = self.response
mod = self.importer.load_module(self.modname)
self.assertTrue(mod.__package__ is None)
self.assertIsNone(mod.__package__)
class LoadSubmoduleTest(ImporterMixin, testlib.TestCase):
@ -96,7 +96,7 @@ class LoadModulePackageTest(ImporterMixin, testlib.TestCase):
def test_module_loader_set(self):
self.context.send_await.return_value = self.response
mod = self.importer.load_module(self.modname)
self.assertTrue(mod.__loader__ is self.importer)
self.assertIs(mod.__loader__, self.importer)
def test_module_path_present(self):
self.context.send_await.return_value = self.response
@ -111,7 +111,7 @@ class LoadModulePackageTest(ImporterMixin, testlib.TestCase):
def test_module_data(self):
self.context.send_await.return_value = self.response
mod = self.importer.load_module(self.modname)
self.assertTrue(isinstance(mod.func, types.FunctionType))
self.assertIsInstance(mod.func, types.FunctionType)
self.assertEquals(mod.func.__module__, self.modname)

7
tests/master_test.py
View File

@ -36,7 +36,7 @@ class IterReadTest(unittest.TestCase):
try:
reader = self.func(proc.stdout.fileno())
for i, chunk in enumerate(reader, 1):
assert i == int(chunk)
self.assertEqual(i, int(chunk))
if i > 3:
break
finally:
@ -52,7 +52,7 @@ class IterReadTest(unittest.TestCase):
got.append(chunk)
assert 0, 'TimeoutError not raised'
except mitogen.core.TimeoutError:
assert len(got) == 0
self.assertEqual(len(got), 0)
finally:
proc.terminate()
@ -68,7 +68,8 @@ class IterReadTest(unittest.TestCase):
except mitogen.core.TimeoutError:
# Give a little wiggle room in case of imperfect scheduling.
# Ideal number should be 9.
assert 3 < len(got) < 5
self.assertLess(3, len(got))
self.assertLess(len(got), 5)
finally:
proc.terminate()

4
tests/module_finder_test.py
View File

@ -97,12 +97,12 @@ class GetModuleViaSysModulesTest(testlib.TestCase):
# _socket comes from a .so
import _socket
tup = self.call('_socket')
self.assertEquals(None, tup)
self.assertIsNone(tup)
def test_builtin_fails(self):
# sys is built-in
tup = self.call('sys')
self.assertEquals(None, tup)
self.assertIsNone(tup)
class ResolveRelPathTest(testlib.TestCase):

2
tests/nested_test.py
View File

@ -13,7 +13,7 @@ class NestedTest(testlib.RouterMixin, testlib.TestCase):
context = self.router.local(via=context, name='local%d' % x)
pid = context.call(os.getpid)
self.assertTrue(isinstance(pid, int))
self.assertIsInstance(pid, int)
if __name__ == '__main__':

4
tests/responder_test.py
View File

@ -52,7 +52,7 @@ class BrokenModulesTest(unittest.TestCase):
call = router.route.mock_calls[0]
msg, = call[1]
self.assertEquals(50, msg.handle)
self.assertTrue(msg.unpickle() is None)
self.assertIsNone(msg.unpickle())
def test_ansible_six_messed_up_path(self):
# The copy of six.py shipped with Ansible appears in a package whose
@ -75,7 +75,7 @@ class BrokenModulesTest(unittest.TestCase):
call = router.route.mock_calls[0]
msg, = call[1]
self.assertEquals(50, msg.handle)
self.assertTrue(isinstance(msg.unpickle(), tuple))
self.assertIsInstance(msg.unpickle(), tuple)
if __name__ == '__main__':

18
tests/ssh_test.py
View File

@ -53,7 +53,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
except mitogen.ssh.PasswordError, e:
pass
assert e[0] == self.stream_class.password_required_msg
self.assertEqual(e[0], self.stream_class.password_required_msg)
def test_password_incorrect(self):
try:
@ -65,7 +65,7 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
except mitogen.ssh.PasswordError, e:
pass
assert e[0] == self.stream_class.password_incorrect_msg
self.assertEqual(e[0], self.stream_class.password_incorrect_msg)
def test_password_specified(self):
context = self.docker_ssh(
@ -73,8 +73,10 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
password='y',
)
sentinel = 'i-am-mitogen-test-docker-image\n'
assert sentinel == context.call(plain_old_module.get_sentinel_value)
self.assertEqual(
'i-am-mitogen-test-docker-image\n',
context.call(plain_old_module.get_sentinel_value),
)
def test_pubkey_required(self):
try:
@ -85,15 +87,17 @@ class SshTest(testlib.DockerMixin, unittest.TestCase):
except mitogen.ssh.PasswordError, e:
pass
assert e[0] == self.stream_class.password_required_msg
self.assertEqual(e[0], self.stream_class.password_required_msg)
def test_pubkey_specified(self):
context = self.docker_ssh(
username='has-sudo-pubkey',
identity_file=testlib.data_path('docker/has-sudo-pubkey.key'),
)
sentinel = 'i-am-mitogen-test-docker-image\n'
assert sentinel == context.call(plain_old_module.get_sentinel_value)
self.assertEqual(
'i-am-mitogen-test-docker-image\n',
context.call(plain_old_module.get_sentinel_value),
)
if __name__ == '__main__':

4
tests/utils_test.py
View File

@ -21,14 +21,14 @@ class RunWithRouterTest(unittest.TestCase):
def test_run_with_broker(self):
router = mitogen.utils.run_with_router(func0)
self.assertTrue(isinstance(router, mitogen.master.Router))
self.assertIsInstance(router, mitogen.master.Router)
self.assertFalse(router.broker._thread.isAlive())
class WithRouterTest(unittest.TestCase):
def test_with_broker(self):
router = func()
self.assertTrue(isinstance(router, mitogen.master.Router))
self.assertIsInstance(router, mitogen.master.Router)
self.assertFalse(router.broker._thread.isAlive())