Method test_repr_lock

Lib/test/_test_multiprocessing.py:1550–1596

(self)

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

        lock = self.Lock()
        self.assertEqual(f'<Lock(owner=None)>', repr(lock))

        lock.acquire()
        self.assertEqual(f'<Lock(owner=MainProcess)>', repr(lock))
        lock.release()

        tname = 'T1'
        l = []
        t = threading.Thread(target=self._acquire,
                             args=(lock, l),
                             name=tname)
        t.start()
        time.sleep(0.1)
        self.assertEqual(f'<Lock(owner=MainProcess|{tname})>', l[0])
        lock.release()

        t = threading.Thread(target=self._acquire,
                             args=(lock,),
                             name=tname)
        t.start()
        time.sleep(0.1)
        self.assertEqual('<Lock(owner=SomeOtherThread)>', repr(lock))
        lock.release()

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

        lock = self.Lock()
        event = self.Event()
        p = self.Process(target=self._acquire_event,
                         args=(lock, event),
                         name='P2')
        p.start()
        event.wait()
        self.assertEqual(f'<Lock(owner=SomeOtherProcess)>', repr(lock))
        p.terminate()

    def test_lock(self):
        lock = self.Lock()