def setUpClass(cls):
   (cls._server_port_1, cls._debug_server_url_1, _, cls._server_thread_1,
    cls._server_1) = grpc_debug_test_server.start_server_on_separate_thread(
        dump_to_filesystem=False)
   (cls._server_port_2, cls._debug_server_url_2, _, cls._server_thread_2,
    cls._server_2) = grpc_debug_test_server.start_server_on_separate_thread(
        dump_to_filesystem=False)

 def setUpClass(cls):
   test_util.TensorFlowTestCase.setUpClass()
   (cls._server_port, cls._debug_server_url, cls._server_dump_dir,
    cls._server_thread,
    cls._server) = grpc_debug_test_server.start_server_on_separate_thread()
   cls._server_address = "localhost:%d" % cls._server_port
   (cls._server_port_2, cls._debug_server_url_2, cls._server_dump_dir_2,
    cls._server_thread_2,
    cls._server_2) = grpc_debug_test_server.start_server_on_separate_thread()
   cls._server_address_2 = "localhost:%d" % cls._server_port_2
   cls._curr_file_path = os.path.normpath(os.path.abspath(__file__))

  def setUpClass(cls):
    gpu_memory_fraction_opt = (
        "--gpu_memory_fraction=%f" % cls.PER_PROC_GPU_MEMORY_FRACTION)

    worker_port = portpicker.pick_unused_port()
    cluster_spec = "worker|localhost:%d" % worker_port
    tf_logging.info("cluster_spec: %s", cluster_spec)

    server_bin = test.test_src_dir_path("python/debug/grpc_tensorflow_server")

    cls.server_target = "grpc://localhost:%d" % worker_port

    cls.server_procs = {}
    cls.server_procs["worker"] = subprocess.Popen(
        [
            server_bin,
            "--cluster_spec=%s" % cluster_spec,
            "--job_name=worker",
            "--task_id=0",
            gpu_memory_fraction_opt,
        ],
        stdout=sys.stdout,
        stderr=sys.stderr)

    # Start debug server in-process, on separate thread.
    (cls.debug_server_port, cls.debug_server_url, _, cls.debug_server_thread,
     cls.debug_server
    ) = grpc_debug_test_server.start_server_on_separate_thread(
        dump_to_filesystem=False)
    tf_logging.info("debug server url: %s", cls.debug_server_url)

    cls.session_config = config_pb2.ConfigProto(
        gpu_options=config_pb2.GPUOptions(
            per_process_gpu_memory_fraction=cls.PER_PROC_GPU_MEMORY_FRACTION))

  def testDebuggedSessionRunWorksWithDelayedDebugServerStartup(self):
    """Test debugged Session.run() tolerates delayed debug server startup."""
    ops.reset_default_graph()

    # Start a debug server asynchronously, with a certain amount of delay.
    (debug_server_port, _, _, server_thread,
     debug_server) = grpc_debug_test_server.start_server_on_separate_thread(
         server_start_delay_sec=2.0, dump_to_filesystem=False)

    with self.test_session() as sess:
      a_init = constant_op.constant(42.0, name="a_init")
      a = variables.Variable(a_init, name="a")

      def watch_fn(fetches, feeds):
        del fetches, feeds
        return framework.WatchOptions(debug_ops=["DebugIdentity"])

      sess = grpc_wrapper.GrpcDebugWrapperSession(
          sess, "localhost:%d" % debug_server_port, watch_fn=watch_fn)
      sess.run(a.initializer)
      self.assertAllClose(
          [42.0], debug_server.debug_tensor_values["a_init:0:DebugIdentity"])

    debug_server.stop_server().wait()
    server_thread.join()

 def testStartServerWithoutBlocking(self):
   (_, _, _, server_thread,
    server) = grpc_debug_test_server.start_server_on_separate_thread(
        poll_server=True, blocking=False)
   # The thread that starts the server shouldn't block, so we should be able to
   # join it before stopping the server.
   server_thread.join()
   server.stop_server().wait()

  def testRunServerAfterStopRaisesException(self):
    (_, _, _, server_thread,
     server) = grpc_debug_test_server.start_server_on_separate_thread(
         poll_server=True)
    server.stop_server().wait()
    server_thread.join()

    with self.assertRaisesRegexp(ValueError, "Server has already stopped"):
      server.run_server()

  def testRepeatedRunServerRaisesException(self):
    (_, _, _, server_thread,
     server) = grpc_debug_test_server.start_server_on_separate_thread(
         poll_server=True)
    # The server is started asynchronously. It needs to be polled till its state
    # has become started.

    with self.assertRaisesRegexp(
        ValueError, "Server has already started running"):
      server.run_server()

    server.stop_server().wait()
    server_thread.join()

  def testToggleWatchesOnCoreMetadata(self):
    (_, debug_server_url, _, server_thread,
     server) = grpc_debug_test_server.start_server_on_separate_thread(
         dump_to_filesystem=False,
         toggle_watch_on_core_metadata=[("toggled_1", 0, "DebugIdentity"),
                                        ("toggled_2", 0, "DebugIdentity")])
    self._servers_and_threads.append((server, server_thread))

    with session.Session(
        config=session_debug_testlib.no_rewrite_session_config()) as sess:
      v_1 = variables.VariableV1(50.0, name="v_1")
      v_2 = variables.VariableV1(-50.0, name="v_1")
      # These two nodes have names that match those in the
      # toggle_watch_on_core_metadata argument used when calling
      # start_server_on_separate_thread().
      toggled_1 = constant_op.constant(5.0, name="toggled_1")
      toggled_2 = constant_op.constant(-5.0, name="toggled_2")
      inc_v_1 = state_ops.assign_add(v_1, toggled_1, name="inc_v_1")
      inc_v_2 = state_ops.assign_add(v_2, toggled_2, name="inc_v_2")

      sess.run([v_1.initializer, v_2.initializer])

      run_metadata = config_pb2.RunMetadata()
      run_options = config_pb2.RunOptions(output_partition_graphs=True)
      debug_utils.watch_graph(
          run_options,
          sess.graph,
          debug_ops=["DebugIdentity(gated_grpc=true)"],
          debug_urls=[debug_server_url])

      for i in xrange(4):
        server.clear_data()

        sess.run([inc_v_1, inc_v_2],
                 options=run_options, run_metadata=run_metadata)

        if i % 2 == 0:
          self.assertEqual(2, len(server.debug_tensor_values))
          self.assertAllClose(
              [5.0],
              server.debug_tensor_values["toggled_1:0:DebugIdentity"])
          self.assertAllClose(
              [-5.0],
              server.debug_tensor_values["toggled_2:0:DebugIdentity"])
        else:
          self.assertEqual(0, len(server.debug_tensor_values))

 def setUpClass(cls):
   (cls.debug_server_port, cls.debug_server_url, _, cls.debug_server_thread,
    cls.debug_server
   ) = grpc_debug_test_server.start_server_on_separate_thread(
       dump_to_filesystem=False)
   tf_logging.info("debug server url: %s", cls.debug_server_url)

 def setUpClass(cls):
   session_debug_testlib.SessionDebugTestBase.setUpClass()
   (cls._server_port, cls._debug_server_url, cls._server_dump_dir,
    cls._server_thread,
    cls._server) = grpc_debug_test_server.start_server_on_separate_thread()