Method test_repr_rlock

Lib/test/_test_multiprocessing.py:1641–1685

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    def test_repr_rlock(self):
        if self.TYPE != 'processes':
            self.skipTest('test not appropriate for {}'.format(self.TYPE))

        lock = self.RLock()
        self.assertEqual('<RLock(None, 0)>', repr(lock))

        n = 3
        for _ in range(n):
            lock.acquire()
        self.assertEqual(f'<RLock(MainProcess, {n})>', repr(lock))
        for _ in range(n):
            lock.release()

        t, l = [], []
        for i in range(n):
            t.append(threading.Thread(target=self._acquire_release,
                                      args=(lock, 0.1, l, i+1),
                                      name=f'T{i+1}'))
            t[-1].start()
        for t_ in t:
            t_.join()
        for i in range(n):
            self.assertIn(f'<RLock(MainProcess|T{i+1}, {i+1})>', l)

        rlock = self.RLock()
        t = threading.Thread(target=rlock.acquire)
        t.start()
        t.join()
        self.assertEqual('<RLock(SomeOtherThread, nonzero)>', repr(rlock))

        pname = 'P1'
        l = multiprocessing.Manager().list()
        p = self.Process(target=self._acquire_release,
                         args=(lock, 0.1, l),
                         name=pname)
        p.start()
        p.join()
        self.assertEqual(f'<RLock({pname}, 1)>', l[0])

        rlock = self.RLock()
        p = self.Process(target=self._acquire, args=(rlock,))
        p.start()
        p.join()
        self.assertEqual('<RLock(SomeOtherProcess, nonzero)>', repr(rlock))

    def test_rlock(self):
        lock = self.RLock()