Method test_remote

Lib/test/_test_multiprocessing.py:3544–3570

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    @support.skip_if_sanitizer('TSan: leaks threads', thread=True)
    def test_remote(self):
        authkey = os.urandom(32)

        manager = QueueManager(
            address=(socket_helper.HOST, 0), authkey=authkey, serializer=SERIALIZER,
            shutdown_timeout=SHUTDOWN_TIMEOUT)
        manager.start()
        self.addCleanup(manager.shutdown)

        p = self.Process(target=self._putter, args=(manager.address, authkey))
        p.daemon = True
        p.start()

        manager2 = QueueManager2(
            address=manager.address, authkey=authkey, serializer=SERIALIZER,
            shutdown_timeout=SHUTDOWN_TIMEOUT)
        manager2.connect()
        queue = manager2.get_queue()

        self.assertEqual(queue.get(), self.result)

        # Because we are using xmlrpclib for serialization instead of
        # pickle this will cause a serialization error.
        self.assertRaises(Exception, queue.put, time.sleep)

        # Make queue finalizer run before the server is stopped
        del queue


@hashlib_helper.requires_hashdigest('sha256')