Method run

tensorboard/util/grpc_util_test.py:53–73

Context manager to run the gRPC server and yield a client for it.

(self)

    @contextlib.contextmanager
    def run(self):
        """Context manager to run the gRPC server and yield a client for it."""
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
        grpc_util_test_pb2_grpc.add_TestServiceServicer_to_server(self, server)
        port = server.add_secure_port(
            "localhost:0", grpc.local_server_credentials()
        )

        def launch_server():
            server.start()
            server.wait_for_termination()

        thread = threading.Thread(target=launch_server, name="TestGrpcServer")
        thread.daemon = True
        thread.start()
        with grpc.secure_channel(
            "localhost:%d" % port, grpc.local_channel_credentials()
        ) as channel:
            yield grpc_util_test_pb2_grpc.TestServiceStub(channel)
        server.stop(grace=None)
        thread.join()


class CallWithRetriesTest(tb_test.TestCase):