def _v1_asset_saved_model(self):
   export_graph = ops.Graph()
   vocab_path = os.path.join(self.get_temp_dir(), "vocab.txt")
   with open(vocab_path, "w") as f:
     f.write("alpha\nbeta\ngamma\n")
   with export_graph.as_default():
     initializer = lookup_ops.TextFileInitializer(
         vocab_path,
         key_dtype=dtypes.string,
         key_index=lookup_ops.TextFileIndex.WHOLE_LINE,
         value_dtype=dtypes.int64,
         value_index=lookup_ops.TextFileIndex.LINE_NUMBER)
     table = lookup_ops.HashTable(
         initializer, default_value=-1)
     start = array_ops.placeholder(
         shape=None, dtype=dtypes.string, name="in")
     output = table.lookup(start, name="out")
     with session_lib.Session() as session:
       session.run([table.initializer])
       path = os.path.join(self.get_temp_dir(), "saved_model", str(ops.uid()))
       simple_save.simple_save(
           session,
           path,
           inputs={"start": start},
           outputs={"output": output},
           legacy_init_op=table.initializer)
   file_io.delete_file(vocab_path)
   return path

 def test_table(self):
   initializer = lookup_ops.TextFileInitializer(
       self._vocab_path,
       key_dtype=dtypes.string,
       key_index=lookup_ops.TextFileIndex.WHOLE_LINE,
       value_dtype=dtypes.int64,
       value_index=lookup_ops.TextFileIndex.LINE_NUMBER)
   root = util.Checkpoint(table=lookup_ops.HashTable(
       initializer, default_value=-1))
   root.table_user = def_function.function(
       root.table.lookup,
       input_signature=[tensor_spec.TensorSpec(None, dtypes.string)])
   self.assertEqual(
       2,
       self.evaluate(root.table_user(constant_op.constant("gamma"))))
   save_dir = os.path.join(self.get_temp_dir(), "saved_model")
   save.save(root, save_dir)
   file_io.delete_file(self._vocab_path)
   self.assertAllClose(
       {"output_0": [2, 0]},
       _import_and_infer(save_dir, {"keys": ["gamma", "alpha"]}))
   second_dir = os.path.join(self.get_temp_dir(), "second_dir")
   # Asset paths should track the location the SavedModel is loaded from.
   file_io.rename(save_dir, second_dir)
   self.assertAllClose(
       {"output_0": [2, 1]},
       _import_and_infer(second_dir, {"keys": ["gamma", "beta"]}))

 def testFileWrite(self):
   file_path = os.path.join(self.get_temp_dir(), "temp_file")
   file_io.write_string_to_file(file_path, "testing")
   self.assertTrue(file_io.file_exists(file_path))
   file_contents = file_io.read_file_to_string(file_path)
   self.assertEqual(b"testing", file_contents)
   file_io.delete_file(file_path)

def create_object_test():
  """Verifies file_io's object manipulation methods ."""
  starttime = int(round(time.time() * 1000))
  dir_name = "%s/tf_gcs_test_%s" % (FLAGS.gcs_bucket_url, starttime)
  print("Creating dir %s." % dir_name)
  file_io.create_dir(dir_name)

  # Create a file in this directory.
  file_name = "%s/test_file.txt" % dir_name
  print("Creating file %s." % file_name)
  file_io.write_string_to_file(file_name, "test file creation.")

  list_files_pattern = "%s/test_file*.txt" % dir_name
  print("Getting files matching pattern %s." % list_files_pattern)
  files_list = file_io.get_matching_files(list_files_pattern)
  print(files_list)

  assert len(files_list) == 1
  assert files_list[0] == file_name

  # Cleanup test files.
  print("Deleting file %s." % file_name)
  file_io.delete_file(file_name)

  # Delete directory.
  print("Deleting directory %s." % dir_name)
  file_io.delete_recursively(dir_name)

    def _tito_out(tensor_in):
      checkpoint_dir = tmp_dir
      if tmp_dir is None:
        checkpoint_dir = tempfile.mkdtemp()

      g = tf.Graph()
      with g.as_default():
        si = tf.placeholder(dtype=tensor_in.dtype, shape=tensor_in.shape, name=tensor_in.op.name)
        so = tito_in(si)
        all_vars = tf.contrib.slim.get_variables_to_restore(exclude=exclude)
        saver = tf.train.Saver(all_vars)
        # Downloading the checkpoint from GCS to local speeds up saver.restore() a lot.
        checkpoint_tmp = os.path.join(checkpoint_dir, 'checkpoint')
        with file_io.FileIO(checkpoint, 'r') as f_in, file_io.FileIO(checkpoint_tmp, 'w') as f_out:
          f_out.write(f_in.read())
        with tf.Session() as sess:
          saver.restore(sess, checkpoint_tmp)
          output_graph_def = tf.graph_util.convert_variables_to_constants(sess,
                                                                          g.as_graph_def(),
                                                                          [so.op.name])
        file_io.delete_file(checkpoint_tmp)
        if tmp_dir is None:
          shutil.rmtree(checkpoint_dir)

      tensors_out = tf.import_graph_def(output_graph_def,
                                        input_map={si.name: tensor_in},
                                        return_elements=[so.name])
      return tensors_out[0]

 def testRename(self):
   file_path = os.path.join(self.get_temp_dir(), "temp_file")
   file_io.write_string_to_file(file_path, "testing")
   rename_path = os.path.join(self.get_temp_dir(), "rename_file")
   file_io.rename(file_path, rename_path)
   self.assertTrue(file_io.file_exists(rename_path))
   self.assertFalse(file_io.file_exists(file_path))
   file_io.delete_file(rename_path)

 def testCopyOverwriteFalse(self):
   file_path = os.path.join(self.get_temp_dir(), "temp_file")
   file_io.write_string_to_file(file_path, "testing")
   copy_path = os.path.join(self.get_temp_dir(), "copy_file")
   file_io.write_string_to_file(copy_path, "copy")
   with self.assertRaises(errors.AlreadyExistsError):
     file_io.copy(file_path, copy_path, overwrite=False)
   file_io.delete_file(file_path)
   file_io.delete_file(copy_path)

 def testCopy(self):
   file_path = os.path.join(self.get_temp_dir(), "temp_file")
   file_io.write_string_to_file(file_path, "testing")
   copy_path = os.path.join(self.get_temp_dir(), "copy_file")
   file_io.copy(file_path, copy_path)
   self.assertTrue(file_io.file_exists(copy_path))
   self.assertEqual(b"testing", file_io.read_file_to_string(file_path))
   file_io.delete_file(file_path)
   file_io.delete_file(copy_path)

 def delete_file(self, filename):
     """Proxy for tensorflow.python.lib.io.file_io.delete_file function. Mocks
     the function if a real GCS bucket is not available for testing.
     """
     if not self.mock_gcs:
         tf_file_io.delete_file(filename)
     elif filename.startswith(self._gcs_prefix):
         self.local_objects.pop(filename)
     else:
         os.remove(filename)

 def testRenameOverwriteFalse(self):
   file_path = os.path.join(self.get_temp_dir(), "temp_file")
   file_io.write_string_to_file(file_path, "testing")
   rename_path = os.path.join(self.get_temp_dir(), "rename_file")
   file_io.write_string_to_file(rename_path, "rename")
   with self.assertRaises(errors.AlreadyExistsError):
     file_io.rename(file_path, rename_path, overwrite=False)
   self.assertTrue(file_io.file_exists(rename_path))
   self.assertTrue(file_io.file_exists(file_path))
   file_io.delete_file(rename_path)
   file_io.delete_file(file_path)

 def testAtomicWriteStringToFileOverwriteFalse(self):
   file_path = os.path.join(self._base_dir, "temp_file")
   file_io.atomic_write_string_to_file(file_path, "old", overwrite=False)
   with self.assertRaises(errors.AlreadyExistsError):
     file_io.atomic_write_string_to_file(file_path, "new", overwrite=False)
   file_contents = file_io.read_file_to_string(file_path)
   self.assertEqual("old", file_contents)
   file_io.delete_file(file_path)
   file_io.atomic_write_string_to_file(file_path, "new", overwrite=False)
   file_contents = file_io.read_file_to_string(file_path)
   self.assertEqual("new", file_contents)

 def test_plot_model_rnn(self):
   model = keras.Sequential()
   model.add(
       keras.layers.LSTM(
           16, return_sequences=True, input_shape=(2, 3), name='lstm'))
   model.add(keras.layers.TimeDistributed(keras.layers.Dense(5, name='dense')))
   dot_img_file = 'model_2.png'
   try:
     vis_utils.plot_model(model, to_file=dot_img_file, show_shapes=True)
     self.assertTrue(file_io.file_exists(dot_img_file))
     file_io.delete_file(dot_img_file)
   except ImportError:
     pass

 def testGetMatchingFiles(self):
   dir_path = os.path.join(self.get_temp_dir(), "temp_dir")
   file_io.create_dir(dir_path)
   files = ["file1.txt", "file2.txt", "file3.txt"]
   for name in files:
     file_path = os.path.join(dir_path, name)
     file_io.write_string_to_file(file_path, "testing")
   expected_match = [os.path.join(dir_path, name) for name in files]
   self.assertItemsEqual(file_io.get_matching_files(os.path.join(dir_path,
                                                                 "file*.txt")),
                         expected_match)
   for name in files:
     file_path = os.path.join(dir_path, name)
     file_io.delete_file(file_path)

 def test_plot_model_cnn(self):
   model = keras.Sequential()
   model.add(
       keras.layers.Conv2D(
           filters=2, kernel_size=(2, 3), input_shape=(3, 5, 5), name='conv'))
   model.add(keras.layers.Flatten(name='flat'))
   model.add(keras.layers.Dense(5, name='dense'))
   dot_img_file = 'model_1.png'
   try:
     vis_utils.plot_model(model, to_file=dot_img_file, show_shapes=True)
     self.assertTrue(file_io.file_exists(dot_img_file))
     file_io.delete_file(dot_img_file)
   except ImportError:
     pass

  def test_assets(self):
    file1 = self._make_asset("contents 1")
    file2 = self._make_asset("contents 2")

    root = tracking.AutoCheckpointable()
    root.asset1 = tracking.TrackableAsset(file1)
    root.asset2 = tracking.TrackableAsset(file2)

    save_dir = os.path.join(self.get_temp_dir(), "save_dir")
    save.save(root, save_dir, signatures={})

    file_io.delete_file(file1)
    file_io.delete_file(file2)
    load_dir = os.path.join(self.get_temp_dir(), "load_dir")
    file_io.rename(save_dir, load_dir)

    imported = load.load(load_dir)
    with open(imported.asset1.asset_path.numpy(), "r") as f:
      self.assertEquals("contents 1", f.read())
    with open(imported.asset2.asset_path.numpy(), "r") as f:
      self.assertEquals("contents 2", f.read())

def _delete_file_if_exists(filespec):
  """Deletes files matching `filespec`."""
  for pathname in file_io.get_matching_files(filespec):
    file_io.delete_file(pathname)

  def testAPIDefCompatibility(self):
    # Get base ApiDef
    name_to_base_api_def = self._GetBaseApiMap()
    # Extract Python API
    visitor = python_object_to_proto_visitor.PythonObjectToProtoVisitor()
    public_api_visitor = public_api.PublicAPIVisitor(visitor)
    public_api_visitor.do_not_descend_map['tf'].append('contrib')
    traverse.traverse(tf, public_api_visitor)
    proto_dict = visitor.GetProtos()

    # Map from first character of op name to Python ApiDefs.
    api_def_map = defaultdict(api_def_pb2.ApiDefs)
    # We need to override all endpoints even if 1 endpoint differs from base
    # ApiDef. So, we first create a map from an op to all its endpoints.
    op_to_endpoint_name = defaultdict(list)

    # Generate map from generated python op to endpoint names.
    for public_module, value in proto_dict.items():
      module_obj = _GetSymbol(public_module)
      for sym in value.tf_module.member_method:
        obj = getattr(module_obj, sym.name)

        # Check if object is defined in gen_* module. That is,
        # the object has been generated from OpDef.
        if hasattr(obj, '__module__') and _IsGenModule(obj.__module__):
          if obj.__name__ not in name_to_base_api_def:
            # Symbol might be defined only in Python and not generated from
            # C++ api.
            continue
          relative_public_module = public_module[len('tensorflow.'):]
          full_name = (relative_public_module + '.' + sym.name
                       if relative_public_module else sym.name)
          op_to_endpoint_name[obj].append(full_name)

    # Generate Python ApiDef overrides.
    for op, endpoint_names in op_to_endpoint_name.items():
      api_def = self._CreatePythonApiDef(
          name_to_base_api_def[op.__name__], endpoint_names)
      if api_def:
        api_defs = api_def_map[op.__name__[0].upper()]
        api_defs.op.extend([api_def])

    for key in _ALPHABET:
      # Get new ApiDef for the given key.
      new_api_defs_str = ''
      if key in api_def_map:
        new_api_defs = api_def_map[key]
        new_api_defs.op.sort(key=attrgetter('graph_op_name'))
        new_api_defs_str = str(new_api_defs)

      # Get current ApiDef for the given key.
      api_defs_file_path = os.path.join(
          _PYTHON_API_DIR, 'api_def_%s.pbtxt' % key)
      old_api_defs_str = ''
      if file_io.file_exists(api_defs_file_path):
        old_api_defs_str = file_io.read_file_to_string(api_defs_file_path)

      if old_api_defs_str == new_api_defs_str:
        continue

      if FLAGS.update_goldens:
        if not new_api_defs_str:
          logging.info('Deleting %s...' % api_defs_file_path)
          file_io.delete_file(api_defs_file_path)
        else:
          logging.info('Updating %s...' % api_defs_file_path)
          file_io.write_string_to_file(api_defs_file_path, new_api_defs_str)
      else:
        self.assertMultiLineEqual(
            old_api_defs_str, new_api_defs_str,
            'To update golden API files, run api_compatibility_test locally '
            'with --update_goldens=True flag.')

 def testFileDeleteFail(self):
   file_path = os.path.join(self._base_dir, "temp_file")
   with self.assertRaises(errors.NotFoundError):
     file_io.delete_file(file_path)

 def testFileDelete(self):
   file_path = os.path.join(self._base_dir, "temp_file")
   file_io.FileIO(file_path, mode="w").write("testing")
   file_io.delete_file(file_path)
   self.assertFalse(file_io.file_exists(file_path))

  def testAPIDefCompatibility(self):
    # Get base ApiDef
    name_to_base_api_def = self._GetBaseApiMap()
    snake_to_camel_graph_op_names = {
        self._GenerateLowerCaseOpName(name): name
        for name in name_to_base_api_def.keys()}
    # Extract Python API
    visitor = python_object_to_proto_visitor.PythonObjectToProtoVisitor()
    public_api_visitor = public_api.PublicAPIVisitor(visitor)
    public_api_visitor.do_not_descend_map['tf'].append('contrib')
    traverse.traverse(tf, public_api_visitor)
    proto_dict = visitor.GetProtos()

    # Map from file path to Python ApiDefs.
    new_api_defs_map = defaultdict(api_def_pb2.ApiDefs)
    # We need to override all endpoints even if 1 endpoint differs from base
    # ApiDef. So, we first create a map from an op to all its endpoints.
    op_to_endpoint_name = defaultdict(list)

    # Generate map from generated python op to endpoint names.
    for public_module, value in proto_dict.items():
      module_obj = _GetSymbol(public_module)
      for sym in value.tf_module.member_method:
        obj = getattr(module_obj, sym.name)

        # Check if object is defined in gen_* module. That is,
        # the object has been generated from OpDef.
        if hasattr(obj, '__module__') and _IsGenModule(obj.__module__):
          if obj.__name__ not in snake_to_camel_graph_op_names:
            # Symbol might be defined only in Python and not generated from
            # C++ api.
            continue
          relative_public_module = public_module[len('tensorflow.'):]
          full_name = (relative_public_module + '.' + sym.name
                       if relative_public_module else sym.name)
          op_to_endpoint_name[obj].append(full_name)

    # Generate Python ApiDef overrides.
    for op, endpoint_names in op_to_endpoint_name.items():
      graph_op_name = snake_to_camel_graph_op_names[op.__name__]
      api_def = self._CreatePythonApiDef(
          name_to_base_api_def[graph_op_name], endpoint_names)

      if api_def:
        file_path = _GetApiDefFilePath(graph_op_name)
        api_defs = new_api_defs_map[file_path]
        api_defs.op.extend([api_def])

    self._AddHiddenOpOverrides(name_to_base_api_def, new_api_defs_map)

    old_api_defs_map = _GetGoldenApiDefs()
    for file_path, new_api_defs in new_api_defs_map.items():
      # Get new ApiDef string.
      new_api_defs_str = str(new_api_defs)

      # Get current ApiDef for the given file.
      old_api_defs_str = (
          old_api_defs_map[file_path] if file_path in old_api_defs_map else '')

      if old_api_defs_str == new_api_defs_str:
        continue

      if FLAGS.update_goldens:
        logging.info('Updating %s...' % file_path)
        file_io.write_string_to_file(file_path, new_api_defs_str)
      else:
        self.assertMultiLineEqual(
            old_api_defs_str, new_api_defs_str,
            'To update golden API files, run api_compatibility_test locally '
            'with --update_goldens=True flag.')

    for file_path in set(old_api_defs_map) - set(new_api_defs_map):
      if FLAGS.update_goldens:
        logging.info('Deleting %s...' % file_path)
        file_io.delete_file(file_path)
      else:
        self.fail(
            '%s file is no longer needed and should be removed.'
            'To update golden API files, run api_compatibility_test locally '
            'with --update_goldens=True flag.' % file_path)