Method test_lock_conflict

Lib/test/test_mailbox.py:1232–1265

(self)

    @support.requires_fork()
    @unittest.skipUnless(hasattr(socket, 'socketpair'), "Test needs socketpair().")
    def test_lock_conflict(self):
        # Fork off a child process that will lock the mailbox temporarily,
        # unlock it and exit.
        c, p = socket.socketpair()
        self.addCleanup(c.close)
        self.addCleanup(p.close)

        pid = os.fork()
        if pid == 0:
            # child
            try:
                # lock the mailbox, and signal the parent it can proceed
                self._box.lock()
                c.send(b'c')

                # wait until the parent is done, and unlock the mailbox
                c.recv(1)
                self._box.unlock()
            finally:
                os._exit(0)

        # In the parent, wait until the child signals it locked the mailbox.
        p.recv(1)
        try:
            self.assertRaises(mailbox.ExternalClashError,
                              self._box.lock)
        finally:
            # Signal the child it can now release the lock and exit.
            p.send(b'p')
            # Wait for child to exit.  Locking should now succeed.
            support.wait_process(pid, exitcode=0)

        self._box.lock()
        self._box.unlock()

    def test_relock(self):
        # Test case for bug #1575506: the mailbox class was locking the