if pid is not None:
os.kill(pid, signal.SIGKILL)
os.waitpid(pid, 0)
- with warnings.catch_warnings(record=True) as all_warn:
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
_semaphore_tracker.ensure_running()
pid = _semaphore_tracker._pid
ctx = multiprocessing.get_context("spawn")
with warnings.catch_warnings(record=True) as all_warn:
+ warnings.simplefilter("always")
sem = ctx.Semaphore()
sem.acquire()
sem.release()
if should_die:
self.assertEqual(len(all_warn), 1)
the_warn = all_warn[0]
- issubclass(the_warn.category, UserWarning)
+ self.assertTrue(issubclass(the_warn.category, UserWarning))
self.assertTrue("semaphore_tracker: process died"
in str(the_warn.message))
else: