tests: Replace uses of assertTrue() with specific methods

pull/949/head
Alex Willmer 2 years ago
parent eb4a7e0ad5
commit e36bbde9ac

@ -42,7 +42,7 @@ class WatcherTest(testlib.TestCase):
self.tf.seek(0) self.tf.seek(0)
self.tf.truncate(0) self.tf.truncate(0)
watcher.check() watcher.check()
self.assertTrue(b('SOMEKEY') not in environb) self.assertNotIn(b('SOMEKEY'), environb)
def test_key_added(self): def test_key_added(self):
watcher = klass(self.tf.name) watcher = klass(self.tf.name)

@ -13,18 +13,18 @@ class ConstructorTest(testlib.TestCase):
def test_string_noargs(self): def test_string_noargs(self):
e = self.klass('%s%s') e = self.klass('%s%s')
self.assertEqual(e.args[0], '%s%s') self.assertEqual(e.args[0], '%s%s')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_string_args(self): def test_string_args(self):
e = self.klass('%s%s', 1, 1) e = self.klass('%s%s', 1, 1)
self.assertEqual(e.args[0], '11') self.assertEqual(e.args[0], '11')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_from_exc(self): def test_from_exc(self):
ve = plain_old_module.MyError('eek') ve = plain_old_module.MyError('eek')
e = self.klass(ve) e = self.klass(ve)
self.assertEqual(e.args[0], 'plain_old_module.MyError: eek') self.assertEqual(e.args[0], 'plain_old_module.MyError: eek')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_form_base_exc(self): def test_form_base_exc(self):
ve = SystemExit('eek') ve = SystemExit('eek')
@ -33,7 +33,7 @@ class ConstructorTest(testlib.TestCase):
self.assertEqual(e.args[0], self.assertEqual(e.args[0],
# varies across 2/3. # varies across 2/3.
'%s.%s: eek' % (cls.__module__, cls.__name__)) '%s.%s: eek' % (cls.__module__, cls.__name__))
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_from_exc_tb(self): def test_from_exc_tb(self):
try: try:
@ -43,18 +43,18 @@ class ConstructorTest(testlib.TestCase):
e = self.klass(ve) e = self.klass(ve)
self.assertTrue(e.args[0].startswith('plain_old_module.MyError: eek')) self.assertTrue(e.args[0].startswith('plain_old_module.MyError: eek'))
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
self.assertTrue('test_from_exc_tb' in e.args[0]) self.assertIn('test_from_exc_tb', e.args[0])
def test_bytestring_conversion(self): def test_bytestring_conversion(self):
e = self.klass(mitogen.core.b('bytes')) e = self.klass(mitogen.core.b('bytes'))
self.assertEqual(u'bytes', e.args[0]) self.assertEqual(u'bytes', e.args[0])
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_reduce(self): def test_reduce(self):
e = self.klass('eek') e = self.klass('eek')
func, (arg,) = e.__reduce__() func, (arg,) = e.__reduce__()
self.assertTrue(func is mitogen.core._unpickle_call_error) self.assertIs(func, mitogen.core._unpickle_call_error)
self.assertEqual(arg, e.args[0]) self.assertEqual(arg, e.args[0])
@ -105,4 +105,4 @@ class PickleTest(testlib.TestCase):
e2 = pickle.loads(pickle.dumps(e)) e2 = pickle.loads(pickle.dumps(e))
self.assertTrue(e2.args[0].startswith('plain_old_module.MyError: eek')) self.assertTrue(e2.args[0].startswith('plain_old_module.MyError: eek'))
self.assertTrue('test_from_exc_tb' in e2.args[0]) self.assertIn('test_from_exc_tb', e2.args[0])

@ -39,7 +39,7 @@ class ConnectionTest(testlib.RouterMixin, testlib.TestCase):
th.join() th.join()
exc, = result exc, = result
self.assertTrue(isinstance(exc, mitogen.parent.CancelledError)) self.assertIsInstance(exc, mitogen.parent.CancelledError)
self.assertEqual(mitogen.parent.BROKER_SHUTDOWN_MSG, exc.args[0]) self.assertEqual(mitogen.parent.BROKER_SHUTDOWN_MSG, exc.args[0])

@ -207,8 +207,7 @@ class TtyCreateChildTest(testlib.TestCase):
self.assertTrue(stat.S_ISCHR(st.st_mode)) self.assertTrue(stat.S_ISCHR(st.st_mode))
self.assertTrue(stat.S_ISCHR(info['st_mode'])) self.assertTrue(stat.S_ISCHR(info['st_mode']))
self.assertTrue(isinstance(info['ttyname'], self.assertIsInstance(info['ttyname'], mitogen.core.UnicodeType)
mitogen.core.UnicodeType))
self.assertTrue(os.isatty(proc.stdin.fileno())) self.assertTrue(os.isatty(proc.stdin.fileno()))
flags = fcntl.fcntl(proc.stdin.fileno(), fcntl.F_GETFL) flags = fcntl.fcntl(proc.stdin.fileno(), fcntl.F_GETFL)
@ -225,8 +224,7 @@ class TtyCreateChildTest(testlib.TestCase):
self.assertTrue(stat.S_ISCHR(st.st_mode)) self.assertTrue(stat.S_ISCHR(st.st_mode))
self.assertTrue(stat.S_ISCHR(info['st_mode'])) self.assertTrue(stat.S_ISCHR(info['st_mode']))
self.assertTrue(isinstance(info['ttyname'], self.assertIsInstance(info['ttyname'], mitogen.core.UnicodeType)
mitogen.core.UnicodeType))
self.assertTrue(os.isatty(proc.stdout.fileno())) self.assertTrue(os.isatty(proc.stdout.fileno()))
flags = fcntl.fcntl(proc.stdout.fileno(), fcntl.F_GETFL) flags = fcntl.fcntl(proc.stdout.fileno(), fcntl.F_GETFL)
@ -247,8 +245,7 @@ class TtyCreateChildTest(testlib.TestCase):
self.assertTrue(stat.S_ISCHR(st.st_mode)) self.assertTrue(stat.S_ISCHR(st.st_mode))
self.assertTrue(stat.S_ISCHR(info['st_mode'])) self.assertTrue(stat.S_ISCHR(info['st_mode']))
self.assertTrue(isinstance(info['ttyname'], self.assertIsInstance(info['ttyname'], mitogen.core.UnicodeType)
mitogen.core.UnicodeType))
self.assertTrue(os.isatty(proc.stdout.fileno())) self.assertTrue(os.isatty(proc.stdout.fileno()))
flags = fcntl.fcntl(proc.stdout.fileno(), fcntl.F_GETFL) flags = fcntl.fcntl(proc.stdout.fileno(), fcntl.F_GETFL)
@ -295,8 +292,7 @@ class StderrDiagTtyMixin(object):
self.assertTrue(stat.S_ISCHR(st.st_mode)) self.assertTrue(stat.S_ISCHR(st.st_mode))
self.assertTrue(stat.S_ISCHR(info['st_mode'])) self.assertTrue(stat.S_ISCHR(info['st_mode']))
self.assertTrue(isinstance(info['ttyname'], self.assertIsInstance(info['ttyname'], mitogen.core.UnicodeType)
mitogen.core.UnicodeType))
self.assertTrue(os.isatty(proc.stderr.fileno())) self.assertTrue(os.isatty(proc.stderr.fileno()))
flags = fcntl.fcntl(proc.stderr.fileno(), fcntl.F_GETFL) flags = fcntl.fcntl(proc.stderr.fileno(), fcntl.F_GETFL)

@ -37,7 +37,7 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# e = self.assertRaises(mitogen.core.StreamError, # e = self.assertRaises(mitogen.core.StreamError,
# lambda: self.router.doas(via=ssh) # lambda: self.router.doas(via=ssh)
# ) # )
# self.assertTrue(mitogen.doas.password_required_msg in str(e)) # self.assertIn(mitogen.doas.password_required_msg, str(e))
# def test_password_incorrect(self): # def test_password_incorrect(self):
# ssh = self.docker_ssh( # ssh = self.docker_ssh(
@ -47,7 +47,7 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# e = self.assertRaises(mitogen.core.StreamError, # e = self.assertRaises(mitogen.core.StreamError,
# lambda: self.router.doas(via=ssh, password='x') # lambda: self.router.doas(via=ssh, password='x')
# ) # )
# self.assertTrue(mitogen.doas.password_incorrect_msg in str(e)) # self.assertIn(mitogen.doas.password_incorrect_msg, str(e))
# def test_password_okay(self): # def test_password_okay(self):
# ssh = self.docker_ssh( # ssh = self.docker_ssh(

@ -8,19 +8,19 @@ class ConstructorTest(testlib.TestCase):
def test_literal_no_format(self): def test_literal_no_format(self):
e = self.klass('error') e = self.klass('error')
self.assertEqual(e.args[0], 'error') self.assertEqual(e.args[0], 'error')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_literal_format_chars_present(self): def test_literal_format_chars_present(self):
e = self.klass('error%s') e = self.klass('error%s')
self.assertEqual(e.args[0], 'error%s') self.assertEqual(e.args[0], 'error%s')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_format(self): def test_format(self):
e = self.klass('error%s', 123) e = self.klass('error%s', 123)
self.assertEqual(e.args[0], 'error123') self.assertEqual(e.args[0], 'error123')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)
def test_bytes_to_unicode(self): def test_bytes_to_unicode(self):
e = self.klass(mitogen.core.b('error')) e = self.klass(mitogen.core.b('error'))
self.assertEqual(e.args[0], 'error') self.assertEqual(e.args[0], 'error')
self.assertTrue(isinstance(e.args[0], mitogen.core.UnicodeType)) self.assertIsInstance(e.args[0], mitogen.core.UnicodeType)

@ -40,7 +40,7 @@ class FetchTest(testlib.RouterMixin, testlib.TestCase):
pool.stop() pool.stop()
expect = service.unregistered_msg % ('/etc/shadow',) expect = service.unregistered_msg % ('/etc/shadow',)
self.assertTrue(expect in e.args[0]) self.assertIn(expect, e.args[0])
if sys.platform == 'darwin': if sys.platform == 'darwin':
ROOT_GROUP = 'wheel' ROOT_GROUP = 'wheel'
@ -48,13 +48,13 @@ class FetchTest(testlib.RouterMixin, testlib.TestCase):
ROOT_GROUP = 'root' ROOT_GROUP = 'root'
def _validate_response(self, resp): def _validate_response(self, resp):
self.assertTrue(isinstance(resp, dict)) self.assertIsInstance(resp, dict)
self.assertEqual('root', resp['owner']) self.assertEqual('root', resp['owner'])
self.assertEqual(self.ROOT_GROUP, resp['group']) self.assertEqual(self.ROOT_GROUP, resp['group'])
self.assertTrue(isinstance(resp['mode'], int)) self.assertIsInstance(resp['mode'], int)
self.assertTrue(isinstance(resp['mtime'], float)) self.assertIsInstance(resp['mtime'], float)
self.assertTrue(isinstance(resp['atime'], float)) self.assertIsInstance(resp['atime'], float)
self.assertTrue(isinstance(resp['size'], int)) self.assertIsInstance(resp['size'], int)
def test_path_authorized(self): def test_path_authorized(self):
recv = mitogen.core.Receiver(self.router) recv = mitogen.core.Receiver(self.router)
@ -117,7 +117,7 @@ class FetchTest(testlib.RouterMixin, testlib.TestCase):
pool.stop() pool.stop()
expect = service.unregistered_msg % (path,) expect = service.unregistered_msg % (path,)
self.assertTrue(expect in e.args[0]) self.assertIn(expect, e.args[0])
def test_prefix_authorized_abspath_good(self): def test_prefix_authorized_abspath_good(self):
l1 = self.router.local() l1 = self.router.local()
@ -144,4 +144,4 @@ class FetchTest(testlib.RouterMixin, testlib.TestCase):
pool.stop() pool.stop()
expect = service.unregistered_msg % (path,) expect = service.unregistered_msg % (path,)
self.assertTrue(expect in e.args[0]) self.assertIn(expect, e.args[0])

@ -198,7 +198,7 @@ class ThreadedCloseTest(testlib.TestCase):
self.join() self.join()
self.assertEqual(self.results, [None]) self.assertEqual(self.results, [None])
for exc in self.excs: for exc in self.excs:
self.assertTrue(isinstance(exc, mitogen.core.LatchError)) self.assertIsInstance(exc, mitogen.core.LatchError)
def test_five_threads(self): def test_five_threads(self):
latch = self.klass() latch = self.klass()
@ -208,4 +208,4 @@ class ThreadedCloseTest(testlib.TestCase):
self.join() self.join()
self.assertEqual(self.results, [None]*5) self.assertEqual(self.results, [None]*5)
for exc in self.excs: for exc in self.excs:
self.assertTrue(isinstance(exc, mitogen.core.LatchError)) self.assertIsInstance(exc, mitogen.core.LatchError)

@ -65,8 +65,8 @@ class StartupTest(testlib.RouterMixin, testlib.TestCase):
c1.shutdown(wait=True) c1.shutdown(wait=True)
logs = log.stop() logs = log.stop()
self.assertTrue('Python version is' in logs) self.assertIn('Python version is', logs)
self.assertTrue('Parent is context 0 (master)' in logs) self.assertIn('Parent is context 0 (master)', logs)
def test_earliest_messages_logged_via(self): def test_earliest_messages_logged_via(self):
c1 = self.router.local(name='c1') c1 = self.router.local(name='c1')
@ -80,10 +80,10 @@ class StartupTest(testlib.RouterMixin, testlib.TestCase):
c2.shutdown(wait=True) c2.shutdown(wait=True)
logs = log.stop() logs = log.stop()
self.assertTrue('Python version is' in logs) self.assertIn('Python version is', logs)
expect = 'Parent is context %s (%s)' % (c1.context_id, 'parent') expect = 'Parent is context %s (%s)' % (c1.context_id, 'parent')
self.assertTrue(expect in logs) self.assertIn(expect, logs)
StartupTest = unittest.skipIf( StartupTest = unittest.skipIf(
condition=sys.version_info < (2, 7) or sys.version_info >= (3, 6), condition=sys.version_info < (2, 7) or sys.version_info >= (3, 6),

@ -26,7 +26,7 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
argv = eval(context.call(os.getenv, 'ORIGINAL_ARGV')) argv = eval(context.call(os.getenv, 'ORIGINAL_ARGV'))
self.assertEqual(argv[0], self.lxc_attach_path) self.assertEqual(argv[0], self.lxc_attach_path)
self.assertTrue('--clear-env' in argv) self.assertIn('--clear-env', argv)
self.assertTrue(has_subseq(argv, ['--name', 'container_name'])) self.assertTrue(has_subseq(argv, ['--name', 'container_name']))
def test_eof(self): def test_eof(self):

@ -49,12 +49,12 @@ class ConstructorTest(testlib.TestCase):
def test_data_default(self): def test_data_default(self):
m = self.klass() m = self.klass()
self.assertEqual(m.data, b('')) self.assertEqual(m.data, b(''))
self.assertTrue(isinstance(m.data, mitogen.core.BytesType)) self.assertIsInstance(m.data, mitogen.core.BytesType)
def test_data_explicit(self): def test_data_explicit(self):
m = self.klass(data=b('asdf')) m = self.klass(data=b('asdf'))
self.assertEqual(m.data, b('asdf')) self.assertEqual(m.data, b('asdf'))
self.assertTrue(isinstance(m.data, mitogen.core.BytesType)) self.assertIsInstance(m.data, mitogen.core.BytesType)
def test_data_hates_unicode(self): def test_data_hates_unicode(self):
self.assertRaises(Exception, self.assertRaises(Exception,
@ -185,9 +185,9 @@ class PickledTest(testlib.TestCase):
d = {1: 2, u'a': 3, b('b'): 4, 'c': {}} d = {1: 2, u'a': 3, b('b'): 4, 'c': {}}
roundtrip = self.roundtrip(d) roundtrip = self.roundtrip(d)
self.assertEqual(d, roundtrip) self.assertEqual(d, roundtrip)
self.assertTrue(isinstance(roundtrip, dict)) self.assertIsInstance(roundtrip, dict)
for k in d: for k in d:
self.assertTrue(isinstance(roundtrip[k], type(d[k]))) self.assertIsInstance(roundtrip[k], type(d[k]))
def test_int(self): def test_int(self):
self.assertEqual(123, self.klass.pickled(123).unpickle()) self.assertEqual(123, self.klass.pickled(123).unpickle())
@ -195,10 +195,10 @@ class PickledTest(testlib.TestCase):
def test_list(self): def test_list(self):
l = [1, u'b', b('c')] l = [1, u'b', b('c')]
roundtrip = self.roundtrip(l) roundtrip = self.roundtrip(l)
self.assertTrue(isinstance(roundtrip, list)) self.assertIsInstance(roundtrip, list)
self.assertEqual(l, roundtrip) self.assertEqual(l, roundtrip)
for k in range(len(l)): for k in range(len(l)):
self.assertTrue(isinstance(roundtrip[k], type(l[k]))) self.assertIsInstance(roundtrip[k], type(l[k]))
@unittest.skipIf(condition=sys.version_info > (3, 0), @unittest.skipIf(condition=sys.version_info > (3, 0),
reason='long missing in >3.x') reason='long missing in >3.x')
@ -206,21 +206,21 @@ class PickledTest(testlib.TestCase):
l = long(0xffffffffffff) l = long(0xffffffffffff)
roundtrip = self.roundtrip(l) roundtrip = self.roundtrip(l)
self.assertEqual(l, roundtrip) self.assertEqual(l, roundtrip)
self.assertTrue(isinstance(roundtrip, long)) self.assertIsInstance(roundtrip, long)
def test_tuple(self): def test_tuple(self):
l = (1, u'b', b('c')) l = (1, u'b', b('c'))
roundtrip = self.roundtrip(l) roundtrip = self.roundtrip(l)
self.assertEqual(l, roundtrip) self.assertEqual(l, roundtrip)
self.assertTrue(isinstance(roundtrip, tuple)) self.assertIsInstance(roundtrip, tuple)
for k in range(len(l)): for k in range(len(l)):
self.assertTrue(isinstance(roundtrip[k], type(l[k]))) self.assertIsInstance(roundtrip[k], type(l[k]))
def test_unicode(self): def test_unicode(self):
u = u'abcd' u = u'abcd'
roundtrip = self.roundtrip(u) roundtrip = self.roundtrip(u)
self.assertEqual(u, roundtrip) self.assertEqual(u, roundtrip)
self.assertTrue(isinstance(roundtrip, mitogen.core.UnicodeType)) self.assertIsInstance(roundtrip, mitogen.core.UnicodeType)
#### custom types. see also: types_test.py, call_error_test.py #### custom types. see also: types_test.py, call_error_test.py
@ -231,25 +231,25 @@ class PickledTest(testlib.TestCase):
def test_blob_nonempty(self): def test_blob_nonempty(self):
v = mitogen.core.Blob(b('dave')) v = mitogen.core.Blob(b('dave'))
roundtrip = self.roundtrip(v) roundtrip = self.roundtrip(v)
self.assertTrue(isinstance(roundtrip, mitogen.core.Blob)) self.assertIsInstance(roundtrip, mitogen.core.Blob)
self.assertEqual(b('dave'), roundtrip) self.assertEqual(b('dave'), roundtrip)
def test_blob_empty(self): def test_blob_empty(self):
v = mitogen.core.Blob(b('')) v = mitogen.core.Blob(b(''))
roundtrip = self.roundtrip(v) roundtrip = self.roundtrip(v)
self.assertTrue(isinstance(roundtrip, mitogen.core.Blob)) self.assertIsInstance(roundtrip, mitogen.core.Blob)
self.assertEqual(b(''), v) self.assertEqual(b(''), v)
def test_secret_nonempty(self): def test_secret_nonempty(self):
s = mitogen.core.Secret(u'dave') s = mitogen.core.Secret(u'dave')
roundtrip = self.roundtrip(s) roundtrip = self.roundtrip(s)
self.assertTrue(isinstance(roundtrip, mitogen.core.Secret)) self.assertIsInstance(roundtrip, mitogen.core.Secret)
self.assertEqual(u'dave', roundtrip) self.assertEqual(u'dave', roundtrip)
def test_secret_empty(self): def test_secret_empty(self):
s = mitogen.core.Secret(u'') s = mitogen.core.Secret(u'')
roundtrip = self.roundtrip(s) roundtrip = self.roundtrip(s)
self.assertTrue(isinstance(roundtrip, mitogen.core.Secret)) self.assertIsInstance(roundtrip, mitogen.core.Secret)
self.assertEqual(u'', roundtrip) self.assertEqual(u'', roundtrip)
def test_call_error(self): def test_call_error(self):
@ -263,7 +263,7 @@ class PickledTest(testlib.TestCase):
try: try:
c = router.context_by_id(1234) c = router.context_by_id(1234)
roundtrip = self.roundtrip(c) roundtrip = self.roundtrip(c)
self.assertTrue(isinstance(roundtrip, mitogen.core.Context)) self.assertIsInstance(roundtrip, mitogen.core.Context)
self.assertEqual(c.context_id, 1234) self.assertEqual(c.context_id, 1234)
finally: finally:
router.broker.shutdown() router.broker.shutdown()
@ -275,7 +275,7 @@ class PickledTest(testlib.TestCase):
recv = mitogen.core.Receiver(router) recv = mitogen.core.Receiver(router)
sender = recv.to_sender() sender = recv.to_sender()
roundtrip = self.roundtrip(sender, router=router) roundtrip = self.roundtrip(sender, router=router)
self.assertTrue(isinstance(roundtrip, mitogen.core.Sender)) self.assertIsInstance(roundtrip, mitogen.core.Sender)
self.assertEqual(roundtrip.context.context_id, mitogen.context_id) self.assertEqual(roundtrip.context.context_id, mitogen.context_id)
self.assertEqual(roundtrip.dst_handle, sender.dst_handle) self.assertEqual(roundtrip.dst_handle, sender.dst_handle)
finally: finally:
@ -314,7 +314,7 @@ class ReplyTest(testlib.TestCase):
my_reply = mitogen.core.Message.pickled(4444) my_reply = mitogen.core.Message.pickled(4444)
msg.reply(my_reply, router=router) msg.reply(my_reply, router=router)
_, (reply,), _ = router.route.mock_calls[0] _, (reply,), _ = router.route.mock_calls[0]
self.assertTrue(my_reply is reply) self.assertIs(my_reply, reply)
self.assertEqual(reply.dst_id, 1234) self.assertEqual(reply.dst_id, 1234)
self.assertEqual(reply.unpickle(), 4444) self.assertEqual(reply.unpickle(), 4444)

@ -25,4 +25,4 @@ class ReceiveOneTest(testlib.TestCase):
self.assertEqual(1, stream.on_disconnect.call_count) self.assertEqual(1, stream.on_disconnect.call_count)
expect = self.klass.corrupt_msg % (stream.name, junk) expect = self.klass.corrupt_msg % (stream.name, junk)
self.assertTrue(expect in capture.raw()) self.assertIn(expect, capture.raw())

@ -66,7 +66,7 @@ class GetMainModuleDefectivePython3x(testlib.TestCase):
import __main__ import __main__
path, source, is_pkg = self.call('__main__') path, source, is_pkg = self.call('__main__')
self.assertTrue(path is not None) self.assertIsNotNone(path)
self.assertTrue(os.path.exists(path)) self.assertTrue(os.path.exists(path))
self.assertEqual(path, __main__.__file__) self.assertEqual(path, __main__.__file__)
fp = open(path, 'rb') fp = open(path, 'rb')

@ -132,7 +132,7 @@ class StreamErrorTest(testlib.RouterMixin, testlib.TestCase):
) )
) )
expect = mitogen.parent.Connection.eof_error_msg expect = mitogen.parent.Connection.eof_error_msg
self.assertTrue(expect in e.args[0]) self.assertIn(expect, e.args[0])
def test_direct_enoent(self): def test_direct_enoent(self):
e = self.assertRaises(mitogen.core.StreamError, e = self.assertRaises(mitogen.core.StreamError,
@ -154,7 +154,7 @@ class StreamErrorTest(testlib.RouterMixin, testlib.TestCase):
) )
) )
s = 'Child start failed: [Errno 2] No such file or directory' s = 'Child start failed: [Errno 2] No such file or directory'
self.assertTrue(s in e.args[0]) self.assertIn(s, e.args[0])
class ContextTest(testlib.RouterMixin, testlib.TestCase): class ContextTest(testlib.RouterMixin, testlib.TestCase):
@ -173,9 +173,9 @@ class OpenPtyTest(testlib.TestCase):
master_fp, slave_fp = self.func() master_fp, slave_fp = self.func()
try: try:
self.assertTrue(master_fp.isatty()) self.assertTrue(master_fp.isatty())
self.assertTrue(isinstance(master_fp, file)) self.assertIsInstance(master_fp, file)
self.assertTrue(slave_fp.isatty()) self.assertTrue(slave_fp.isatty())
self.assertTrue(isinstance(slave_fp, file)) self.assertIsInstance(slave_fp, file)
finally: finally:
master_fp.close() master_fp.close()
slave_fp.close() slave_fp.close()

@ -163,12 +163,12 @@ class PollMixin(PollerMixin):
def test_empty_zero_timeout(self): def test_empty_zero_timeout(self):
t0 = mitogen.core.now() t0 = mitogen.core.now()
self.assertEqual([], list(self.p.poll(0))) self.assertEqual([], list(self.p.poll(0)))
self.assertTrue((mitogen.core.now() - t0) < .1) # vaguely reasonable self.assertLess((mitogen.core.now() - t0), .1) # vaguely reasonable
def test_empty_small_timeout(self): def test_empty_small_timeout(self):
t0 = mitogen.core.now() t0 = mitogen.core.now()
self.assertEqual([], list(self.p.poll(.2))) self.assertEqual([], list(self.p.poll(.2)))
self.assertTrue((mitogen.core.now() - t0) >= .2) self.assertGreaterEqual((mitogen.core.now() - t0), .2)
class ReadableMixin(PollerMixin, SockMixin): class ReadableMixin(PollerMixin, SockMixin):

@ -9,27 +9,27 @@ class BytesPartitionTest(testlib.TestCase):
def test_no_sep(self): def test_no_sep(self):
left, sep, right = self.func(b('dave'), b('x')) left, sep, right = self.func(b('dave'), b('x'))
self.assertTrue(isinstance(left, mitogen.core.BytesType)) self.assertIsInstance(left, mitogen.core.BytesType)
self.assertTrue(isinstance(sep, mitogen.core.BytesType)) self.assertIsInstance(sep, mitogen.core.BytesType)
self.assertTrue(isinstance(right, mitogen.core.BytesType)) self.assertIsInstance(right, mitogen.core.BytesType)
self.assertEqual(left, b('dave')) self.assertEqual(left, b('dave'))
self.assertEqual(sep, b('')) self.assertEqual(sep, b(''))
self.assertEqual(right, b('')) self.assertEqual(right, b(''))
def test_one_sep(self): def test_one_sep(self):
left, sep, right = self.func(b('davexdave'), b('x')) left, sep, right = self.func(b('davexdave'), b('x'))
self.assertTrue(isinstance(left, mitogen.core.BytesType)) self.assertIsInstance(left, mitogen.core.BytesType)
self.assertTrue(isinstance(sep, mitogen.core.BytesType)) self.assertIsInstance(sep, mitogen.core.BytesType)
self.assertTrue(isinstance(right, mitogen.core.BytesType)) self.assertIsInstance(right, mitogen.core.BytesType)
self.assertEqual(left, b('dave')) self.assertEqual(left, b('dave'))
self.assertEqual(sep, b('x')) self.assertEqual(sep, b('x'))
self.assertEqual(right, b('dave')) self.assertEqual(right, b('dave'))
def test_two_seps(self): def test_two_seps(self):
left, sep, right = self.func(b('davexdavexdave'), b('x')) left, sep, right = self.func(b('davexdavexdave'), b('x'))
self.assertTrue(isinstance(left, mitogen.core.BytesType)) self.assertIsInstance(left, mitogen.core.BytesType)
self.assertTrue(isinstance(sep, mitogen.core.BytesType)) self.assertIsInstance(sep, mitogen.core.BytesType)
self.assertTrue(isinstance(right, mitogen.core.BytesType)) self.assertIsInstance(right, mitogen.core.BytesType)
self.assertEqual(left, b('dave')) self.assertEqual(left, b('dave'))
self.assertEqual(sep, b('x')) self.assertEqual(sep, b('x'))
self.assertEqual(right, b('davexdave')) self.assertEqual(right, b('davexdave'))
@ -40,27 +40,27 @@ class StrPartitionTest(testlib.TestCase):
def test_no_sep(self): def test_no_sep(self):
left, sep, right = self.func(u'dave', u'x') left, sep, right = self.func(u'dave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'dave') self.assertEqual(left, u'dave')
self.assertEqual(sep, u'') self.assertEqual(sep, u'')
self.assertEqual(right, u'') self.assertEqual(right, u'')
def test_one_sep(self): def test_one_sep(self):
left, sep, right = self.func(u'davexdave', u'x') left, sep, right = self.func(u'davexdave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'dave') self.assertEqual(left, u'dave')
self.assertEqual(sep, u'x') self.assertEqual(sep, u'x')
self.assertEqual(right, u'dave') self.assertEqual(right, u'dave')
def test_two_seps(self): def test_two_seps(self):
left, sep, right = self.func(u'davexdavexdave', u'x') left, sep, right = self.func(u'davexdavexdave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'dave') self.assertEqual(left, u'dave')
self.assertEqual(sep, u'x') self.assertEqual(sep, u'x')
self.assertEqual(right, u'davexdave') self.assertEqual(right, u'davexdave')
@ -71,27 +71,27 @@ class StrRpartitionTest(testlib.TestCase):
def test_no_sep(self): def test_no_sep(self):
left, sep, right = self.func(u'dave', u'x') left, sep, right = self.func(u'dave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'') self.assertEqual(left, u'')
self.assertEqual(sep, u'') self.assertEqual(sep, u'')
self.assertEqual(right, u'dave') self.assertEqual(right, u'dave')
def test_one_sep(self): def test_one_sep(self):
left, sep, right = self.func(u'davexdave', u'x') left, sep, right = self.func(u'davexdave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'dave') self.assertEqual(left, u'dave')
self.assertEqual(sep, u'x') self.assertEqual(sep, u'x')
self.assertEqual(right, u'dave') self.assertEqual(right, u'dave')
def test_two_seps(self): def test_two_seps(self):
left, sep, right = self.func(u'davexdavexdave', u'x') left, sep, right = self.func(u'davexdavexdave', u'x')
self.assertTrue(isinstance(left, mitogen.core.UnicodeType)) self.assertIsInstance(left, mitogen.core.UnicodeType)
self.assertTrue(isinstance(sep, mitogen.core.UnicodeType)) self.assertIsInstance(sep, mitogen.core.UnicodeType)
self.assertTrue(isinstance(right, mitogen.core.UnicodeType)) self.assertIsInstance(right, mitogen.core.UnicodeType)
self.assertEqual(left, u'davexdave') self.assertEqual(left, u'davexdave')
self.assertEqual(sep, u'x') self.assertEqual(sep, u'x')
self.assertEqual(right, u'dave') self.assertEqual(right, u'dave')

@ -16,8 +16,8 @@ def yield_stuff_then_die(sender):
class ConstructorTest(testlib.RouterMixin, testlib.TestCase): class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
def test_handle(self): def test_handle(self):
recv = mitogen.core.Receiver(self.router) recv = mitogen.core.Receiver(self.router)
self.assertTrue(isinstance(recv.handle, int)) self.assertIsInstance(recv.handle, int)
self.assertTrue(recv.handle > 100) self.assertGreater(recv.handle, 100)
self.router.route( self.router.route(
mitogen.core.Message.pickled( mitogen.core.Message.pickled(
'hi', 'hi',

@ -25,7 +25,7 @@ class NeutralizeMainTest(testlib.RouterMixin, testlib.TestCase):
_, stderr = proc.communicate() _, stderr = proc.communicate()
self.assertEqual(1, proc.returncode) self.assertEqual(1, proc.returncode)
expect = self.klass.main_guard_msg % (path,) expect = self.klass.main_guard_msg % (path,)
self.assertTrue(expect in stderr.decode()) self.assertIn(expect, stderr.decode())
HAS_MITOGEN_MAIN = mitogen.core.b( HAS_MITOGEN_MAIN = mitogen.core.b(
textwrap.dedent(""" textwrap.dedent("""
@ -181,7 +181,7 @@ class ForwardTest(testlib.RouterMixin, testlib.TestCase):
) )
) )
s = capture.stop() s = capture.stop()
self.assertTrue('dropping forward of' in s) self.assertIn('dropping forward of', s)
def test_stats(self): def test_stats(self):
# Forwarding stats broken because forwarding is broken. See #469. # Forwarding stats broken because forwarding is broken. See #469.

@ -71,7 +71,7 @@ class SourceVerifyTest(testlib.RouterMixin, testlib.TestCase):
# Ensure error was logged. # Ensure error was logged.
expect = 'bad auth_id: got %r via' % (self.child2_msg.auth_id,) expect = 'bad auth_id: got %r via' % (self.child2_msg.auth_id,)
self.assertTrue(expect in log.stop()) self.assertIn(expect, log.stop())
def test_parent_unaware_of_disconnect(self): def test_parent_unaware_of_disconnect(self):
# Parent -> Child A -> Child B. B disconnects concurrent to Parent # Parent -> Child A -> Child B. B disconnects concurrent to Parent
@ -125,7 +125,7 @@ class SourceVerifyTest(testlib.RouterMixin, testlib.TestCase):
# Ensure error was lgoged. # Ensure error was lgoged.
expect = 'bad src_id: got %d via' % (self.child1_msg.src_id,) expect = 'bad src_id: got %d via' % (self.child1_msg.src_id,)
self.assertTrue(expect in log.stop()) self.assertIn(expect, log.stop())
class PolicyTest(testlib.RouterMixin, testlib.TestCase): class PolicyTest(testlib.RouterMixin, testlib.TestCase):
@ -165,7 +165,7 @@ class PolicyTest(testlib.RouterMixin, testlib.TestCase):
self.sync_with_broker() self.sync_with_broker()
# Verify log. # Verify log.
self.assertTrue(self.router.refused_msg in log.stop()) self.assertIn(self.router.refused_msg, log.stop())
# Verify message was not delivered. # Verify message was not delivered.
self.assertTrue(recv.empty()) self.assertTrue(recv.empty())
@ -202,7 +202,7 @@ class CrashTest(testlib.BrokerMixin, testlib.TestCase):
# Ensure it was logged. # Ensure it was logged.
expect = 'broker crashed' expect = 'broker crashed'
self.assertTrue(expect in log.stop()) self.assertIn(expect, log.stop())
self.broker.join() self.broker.join()
@ -264,7 +264,7 @@ class MessageSizeTest(testlib.BrokerMixin, testlib.TestCase):
router.broker.defer_sync(lambda: None) router.broker.defer_sync(lambda: None)
expect = 'message too large (max 4096 bytes)' expect = 'message too large (max 4096 bytes)'
self.assertTrue(expect in logs.stop()) self.assertIn(expect, logs.stop())
def test_local_dead_message(self): def test_local_dead_message(self):
# Local router should generate dead message when reply_to is set. # Local router should generate dead message when reply_to is set.
@ -282,7 +282,7 @@ class MessageSizeTest(testlib.BrokerMixin, testlib.TestCase):
lambda: child.call(zlib.crc32, ' '*8192)) lambda: child.call(zlib.crc32, ' '*8192))
self.assertEqual(e.args[0], expect) self.assertEqual(e.args[0], expect)
self.assertTrue(expect in logs.stop()) self.assertIn(expect, logs.stop())
def test_remote_dead_message(self): def test_remote_dead_message(self):
# Router should send dead message to original recipient when reply_to # Router should send dead message to original recipient when reply_to
@ -325,7 +325,7 @@ class MessageSizeTest(testlib.BrokerMixin, testlib.TestCase):
remote.call(send_n_sized_reply, recv.to_sender(), 128*1024) remote.call(send_n_sized_reply, recv.to_sender(), 128*1024)
expect = 'message too large (max %d bytes)' % (64*1024,) expect = 'message too large (max %d bytes)' % (64*1024,)
self.assertTrue(expect in logs.stop()) self.assertIn(expect, logs.stop())
class NoRouteTest(testlib.RouterMixin, testlib.TestCase): class NoRouteTest(testlib.RouterMixin, testlib.TestCase):

@ -83,14 +83,14 @@ class ActivationTest(testlib.RouterMixin, testlib.TestCase):
l1 = self.router.local() l1 = self.router.local()
counter, id_ = l1.call_service(MyService, 'get_id') counter, id_ = l1.call_service(MyService, 'get_id')
self.assertEqual(1, counter) self.assertEqual(1, counter)
self.assertTrue(isinstance(id_, int)) self.assertIsInstance(id_, int)
def test_sibling_cannot_activate_framework(self): def test_sibling_cannot_activate_framework(self):
l1 = self.router.local(name='l1') l1 = self.router.local(name='l1')
l2 = self.router.local(name='l2') l2 = self.router.local(name='l2')
exc = self.assertRaises(mitogen.core.CallError, exc = self.assertRaises(mitogen.core.CallError,
lambda: l2.call(call_service_in, l1, MyService2.name(), 'get_id')) lambda: l2.call(call_service_in, l1, MyService2.name(), 'get_id'))
self.assertTrue(mitogen.core.Router.refused_msg in exc.args[0]) self.assertIn(mitogen.core.Router.refused_msg, exc.args[0])
def test_sibling_cannot_activate_service(self): def test_sibling_cannot_activate_service(self):
l1 = self.router.local() l1 = self.router.local()
@ -104,7 +104,7 @@ class ActivationTest(testlib.RouterMixin, testlib.TestCase):
finally: finally:
capture.stop() capture.stop()
msg = mitogen.service.Activator.not_active_msg % (MyService2.name(),) msg = mitogen.service.Activator.not_active_msg % (MyService2.name(),)
self.assertTrue(msg in exc.args[0]) self.assertIn(msg, exc.args[0])
def test_activates_only_once(self): def test_activates_only_once(self):
l1 = self.router.local() l1 = self.router.local()
@ -138,7 +138,7 @@ class PermissionTest(testlib.RouterMixin, testlib.TestCase):
u'privileged_op', u'privileged_op',
MyService.name(), MyService.name(),
) )
self.assertTrue(msg in exc.args[0]) self.assertIn(msg, exc.args[0])
class CloseTest(testlib.RouterMixin, testlib.TestCase): class CloseTest(testlib.RouterMixin, testlib.TestCase):

@ -52,7 +52,7 @@ class SshTest(testlib.DockerMixin, testlib.TestCase):
s = capture.stop() s = capture.stop()
expect = "%s: debug1: Reading configuration data" % (context.name,) expect = "%s: debug1: Reading configuration data" % (context.name,)
self.assertTrue(expect in s) self.assertIn(expect, s)
def test_bash_permission_denied(self): def test_bash_permission_denied(self):
# issue #271: only match Permission Denied at start of line. # issue #271: only match Permission Denied at start of line.
@ -161,7 +161,7 @@ class SshTest(testlib.DockerMixin, testlib.TestCase):
fp.seek(0) fp.seek(0)
# Lame test, but we're about to use enforce mode anyway, which # Lame test, but we're about to use enforce mode anyway, which
# verifies the file contents. # verifies the file contents.
self.assertTrue(len(fp.read()) > 0) self.assertGreater(len(fp.read()), 0)
context = self.docker_ssh( context = self.docker_ssh(
username='mitogen__has_sudo', username='mitogen__has_sudo',

@ -45,7 +45,7 @@ class SuTest(testlib.DockerMixin, testlib.TestCase):
e = self.assertRaises(mitogen.core.StreamError, e = self.assertRaises(mitogen.core.StreamError,
lambda: self.router.su(via=ssh) lambda: self.router.su(via=ssh)
) )
self.assertTrue(mitogen.su.password_required_msg in str(e)) self.assertIn(mitogen.su.password_required_msg, str(e))
def test_password_incorrect(self): def test_password_incorrect(self):
ssh = self.docker_ssh( ssh = self.docker_ssh(
@ -55,7 +55,7 @@ class SuTest(testlib.DockerMixin, testlib.TestCase):
e = self.assertRaises(mitogen.core.StreamError, e = self.assertRaises(mitogen.core.StreamError,
lambda: self.router.su(via=ssh, password='x') lambda: self.router.su(via=ssh, password='x')
) )
self.assertTrue(mitogen.su.password_incorrect_msg in str(e)) self.assertIn(mitogen.su.password_incorrect_msg, str(e))
def test_password_okay(self): def test_password_okay(self):
ssh = self.docker_ssh( ssh = self.docker_ssh(

@ -73,7 +73,7 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# e = self.assertRaises(mitogen.core.StreamError, # e = self.assertRaises(mitogen.core.StreamError,
# lambda: self.router.sudo(via=ssh) # lambda: self.router.sudo(via=ssh)
# ) # )
# self.assertTrue(mitogen.sudo.password_required_msg in str(e)) # self.assertIn(mitogen.sudo.password_required_msg, str(e))
# def test_password_incorrect(self): # def test_password_incorrect(self):
# ssh = self.docker_ssh( # ssh = self.docker_ssh(
@ -85,7 +85,7 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# e = self.assertRaises(mitogen.core.StreamError, # e = self.assertRaises(mitogen.core.StreamError,
# lambda: self.router.sudo(via=ssh, password='x') # lambda: self.router.sudo(via=ssh, password='x')
# ) # )
# self.assertTrue(mitogen.sudo.password_incorrect_msg in str(e)) # self.assertIn(mitogen.sudo.password_incorrect_msg, str(e))
# def test_password_okay(self): # def test_password_okay(self):
# ssh = self.docker_ssh( # ssh = self.docker_ssh(
@ -97,4 +97,4 @@ class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
# e = self.assertRaises(mitogen.core.StreamError, # e = self.assertRaises(mitogen.core.StreamError,
# lambda: self.router.sudo(via=ssh, password='rootpassword') # lambda: self.router.sudo(via=ssh, password='rootpassword')
# ) # )
# self.assertTrue(mitogen.sudo.password_incorrect_msg in str(e)) # self.assertIn(mitogen.sudo.password_incorrect_msg, str(e))

@ -87,7 +87,7 @@ class ScheduleTest(TimerListMixin, testlib.TestCase):
timer2 = self.list.schedule(31, lambda: None) timer2 = self.list.schedule(31, lambda: None)
self.assertEqual(31, timer.when) self.assertEqual(31, timer.when)
self.assertEqual(31, timer2.when) self.assertEqual(31, timer2.when)
self.assertTrue(timer is not timer2) self.assertIsNot(timer, timer2)
self.assertEqual(1, self.list.get_timeout()) self.assertEqual(1, self.list.get_timeout())

@ -90,8 +90,8 @@ class KwargsTest(testlib.TestCase):
self.assertEqual({}, kw) self.assertEqual({}, kw)
self.assertEqual('Kwargs({})', repr(kw)) self.assertEqual('Kwargs({})', repr(kw))
klass, (dct,) = kw.__reduce__() klass, (dct,) = kw.__reduce__()
self.assertTrue(klass is self.klass) self.assertIs(klass, self.klass)
self.assertTrue(type(dct) is dict) self.assertIs(type(dct), dict)
self.assertEqual({}, dct) self.assertEqual({}, dct)
@unittest.skipIf(condition=(sys.version_info >= (2, 6)), @unittest.skipIf(condition=(sys.version_info >= (2, 6)),
@ -108,11 +108,11 @@ class KwargsTest(testlib.TestCase):
self.assertEqual({u'key': 123}, kw) self.assertEqual({u'key': 123}, kw)
self.assertEqual("Kwargs({'key': 123})", repr(kw)) self.assertEqual("Kwargs({'key': 123})", repr(kw))
klass, (dct,) = kw.__reduce__() klass, (dct,) = kw.__reduce__()
self.assertTrue(klass is self.klass) self.assertIs(klass, self.klass)
self.assertTrue(type(dct) is dict) self.assertIs(type(dct), dict)
self.assertEqual({u'key': 123}, dct) self.assertEqual({u'key': 123}, dct)
key, = dct key, = dct
self.assertTrue(type(key) is mitogen.core.UnicodeType) self.assertIs(type(key), mitogen.core.UnicodeType)
class AdornedUnicode(mitogen.core.UnicodeType): class AdornedUnicode(mitogen.core.UnicodeType):

Loading…
Cancel
Save