Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-11 19:12:05 +00:00
Code Issues Projects Releases Packages Wiki Activity

gh-71339: Use new assertion methods in the multiprocessing tests (GH-128847)

Browse source
This commit is contained in:
Serhiy Storchaka 2025-01-15 01:17:11 +02:00 • committed by GitHub
parent f7ceb317ae
commit b52de22ac3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 24 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

53
Lib/test/_test_multiprocessing.py
View file

@ -319,7 +319,7 @@ def test_current(self):
authkey = current.authkey authkey = current.authkey
self.assertTrue(current.is_alive()) self.assertTrue(current.is_alive())
self.assertTrue(not current.daemon) self.assertFalse(current.daemon)
self.assertIsInstance(authkey, bytes) self.assertIsInstance(authkey, bytes)
self.assertTrue(len(authkey) > 0) self.assertTrue(len(authkey) > 0)
self.assertEqual(current.ident, os.getpid()) self.assertEqual(current.ident, os.getpid())
@ -463,7 +463,7 @@ def test_process(self):
self.assertEqual(p.is_alive(), False) self.assertEqual(p.is_alive(), False)
self.assertEqual(p.daemon, True) self.assertEqual(p.daemon, True)
self.assertNotIn(p, self.active_children()) self.assertNotIn(p, self.active_children())
self.assertTrue(type(self.active_children()) is list) self.assertIs(type(self.active_children()), list)
self.assertEqual(p.exitcode, None) self.assertEqual(p.exitcode, None)
p.start() p.start()
@ -583,8 +583,8 @@ def test_cpu_count(self):
cpus = multiprocessing.cpu_count() cpus = multiprocessing.cpu_count()
except NotImplementedError: except NotImplementedError:
cpus = 1 cpus = 1
self.assertTrue(type(cpus) is int) self.assertIsInstance(cpus, int)
self.assertTrue(cpus >= 1) self.assertGreaterEqual(cpus, 1)
def test_active_children(self): def test_active_children(self):
self.assertEqual(type(self.active_children()), list) self.assertEqual(type(self.active_children()), list)
@ -2382,14 +2382,14 @@ def test_getobj_getlock(self):
self.assertEqual(lock, lock3) self.assertEqual(lock, lock3)
arr4 = self.Value('i', 5, lock=False) arr4 = self.Value('i', 5, lock=False)
self.assertFalse(hasattr(arr4, 'get_lock')) self.assertNotHasAttr(arr4, 'get_lock')
self.assertFalse(hasattr(arr4, 'get_obj')) self.assertNotHasAttr(arr4, 'get_obj')
self.assertRaises(AttributeError, self.Value, 'i', 5, lock='navalue') self.assertRaises(AttributeError, self.Value, 'i', 5, lock='navalue')
arr5 = self.RawValue('i', 5) arr5 = self.RawValue('i', 5)
self.assertFalse(hasattr(arr5, 'get_lock')) self.assertNotHasAttr(arr5, 'get_lock')
self.assertFalse(hasattr(arr5, 'get_obj')) self.assertNotHasAttr(arr5, 'get_obj')
class _TestArray(BaseTestCase): class _TestArray(BaseTestCase):
@ -2462,14 +2462,14 @@ def test_getobj_getlock_obj(self):
self.assertEqual(lock, lock3) self.assertEqual(lock, lock3)
arr4 = self.Array('i', range(10), lock=False) arr4 = self.Array('i', range(10), lock=False)
self.assertFalse(hasattr(arr4, 'get_lock')) self.assertNotHasAttr(arr4, 'get_lock')
self.assertFalse(hasattr(arr4, 'get_obj')) self.assertNotHasAttr(arr4, 'get_obj')
self.assertRaises(AttributeError, self.assertRaises(AttributeError,
self.Array, 'i', range(10), lock='notalock') self.Array, 'i', range(10), lock='notalock')
arr5 = self.RawArray('i', range(10)) arr5 = self.RawArray('i', range(10))
self.assertFalse(hasattr(arr5, 'get_lock')) self.assertNotHasAttr(arr5, 'get_lock')
self.assertFalse(hasattr(arr5, 'get_obj')) self.assertNotHasAttr(arr5, 'get_obj')
# #
# #
@ -2657,8 +2657,8 @@ def test_namespace(self):
self.assertEqual((n.name, n.job), ('Bob', 'Builder')) self.assertEqual((n.name, n.job), ('Bob', 'Builder'))
del n.job del n.job
self.assertEqual(str(n), "Namespace(name='Bob')") self.assertEqual(str(n), "Namespace(name='Bob')")
self.assertTrue(hasattr(n, 'name')) self.assertHasAttr(n, 'name')
self.assertTrue(not hasattr(n, 'job')) self.assertNotHasAttr(n, 'job')
# #
# #
@ -4938,13 +4938,9 @@ def test_import(self):
for name in modules: for name in modules:
__import__(name) __import__(name)
mod = sys.modules[name] mod = sys.modules[name]
self.assertTrue(hasattr(mod, '__all__'), name) self.assertHasAttr(mod, '__all__', name)
for attr in mod.__all__: for attr in mod.__all__:
self.assertTrue( self.assertHasAttr(mod, attr)
hasattr(mod, attr),
'%r does not have attribute %r' % (mod, attr)
)
# #
# Quick test that logging works -- does not test logging output # Quick test that logging works -- does not test logging output
@ -4957,7 +4953,7 @@ class _TestLogging(BaseTestCase):
def test_enable_logging(self): def test_enable_logging(self):
logger = multiprocessing.get_logger() logger = multiprocessing.get_logger()
logger.setLevel(util.SUBWARNING) logger.setLevel(util.SUBWARNING)
self.assertTrue(logger is not None) self.assertIsNotNone(logger)
logger.debug('this will not be printed') logger.debug('this will not be printed')
logger.info('nor will this') logger.info('nor will this')
logger.setLevel(LOG_LEVEL) logger.setLevel(LOG_LEVEL)
@ -5753,9 +5749,8 @@ def test_set_get(self):
self.assertEqual(multiprocessing.get_start_method(), method) self.assertEqual(multiprocessing.get_start_method(), method)
ctx = multiprocessing.get_context() ctx = multiprocessing.get_context()
self.assertEqual(ctx.get_start_method(), method) self.assertEqual(ctx.get_start_method(), method)
self.assertTrue(type(ctx).__name__.lower().startswith(method)) self.assertStartsWith(type(ctx).__name__.lower(), method)
self.assertTrue( self.assertStartsWith(ctx.Process.__name__.lower(), method)
ctx.Process.__name__.lower().startswith(method))
self.check_context(multiprocessing) self.check_context(multiprocessing)
count += 1 count += 1
finally: finally:
@ -5956,9 +5951,9 @@ def check_resource_tracker_death(self, signum, should_die):
if should_die: if should_die:
self.assertEqual(len(all_warn), 1) self.assertEqual(len(all_warn), 1)
the_warn = all_warn[0] the_warn = all_warn[0]
self.assertTrue(issubclass(the_warn.category, UserWarning)) self.assertIsSubclass(the_warn.category, UserWarning)
self.assertTrue("resource_tracker: process died" self.assertIn("resource_tracker: process died",
in str(the_warn.message)) str(the_warn.message))
else: else:
self.assertEqual(len(all_warn), 0) self.assertEqual(len(all_warn), 0)
@ -6163,8 +6158,8 @@ def is_alive(self):
Process=FailingForkProcess)) Process=FailingForkProcess))
p.close() p.close()
p.join() p.join()
self.assertFalse( for process in forked_processes:
any(process.is_alive() for process in forked_processes)) self.assertFalse(process.is_alive(), process)
@hashlib_helper.requires_hashdigest('sha256') @hashlib_helper.requires_hashdigest('sha256')