Method check_invariant

Lib/test/_test_multiprocessing.py:1801–1810

(self, cond)

        self.assertReturnsIfImplemented(value, func)

    def check_invariant(self, cond):
        # this is only supposed to succeed when there are no sleepers
        if self.TYPE == 'processes':
            try:
                sleepers = (cond._sleeping_count.get_value() -
                            cond._woken_count.get_value())
                self.assertEqual(sleepers, 0)
                self.assertEqual(cond._wait_semaphore.get_value(), 0)
            except NotImplementedError:
                pass

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    def test_notify(self):