aboutsummaryrefslogtreecommitdiffhomepage
path: root/tensorflow/python/platform/default/gfile_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'tensorflow/python/platform/default/gfile_test.py')
-rw-r--r--tensorflow/python/platform/default/gfile_test.py248
1 files changed, 0 insertions, 248 deletions
diff --git a/tensorflow/python/platform/default/gfile_test.py b/tensorflow/python/platform/default/gfile_test.py
deleted file mode 100644
index 38b8b1d30a..0000000000
--- a/tensorflow/python/platform/default/gfile_test.py
+++ /dev/null
@@ -1,248 +0,0 @@
-# Copyright 2015 Google Inc. All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# ==============================================================================
-
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-import contextlib
-import os
-import shutil
-import time
-
-from tensorflow.python.platform.default import _gfile as gfile
-from tensorflow.python.platform.default import _googletest as googletest
-from tensorflow.python.platform.default import _logging as logging
-
-
-class _BaseTest(object):
-
- @property
- def tmp(self):
- return self._tmp_dir
-
- def setUp(self):
- self._orig_dir = os.getcwd()
- self._tmp_dir = googletest.GetTempDir() + "/"
- try:
- os.makedirs(self._tmp_dir)
- except OSError:
- pass # Directory already exists
-
- def tearDown(self):
- try:
- shutil.rmtree(self._tmp_dir)
- except OSError:
- logging.warn("[%s] Post-test directory cleanup failed: %s",
- self, self._tmp_dir)
-
-
-class _GFileBaseTest(_BaseTest):
-
- @property
- def gfile(self):
- raise NotImplementedError("Do not use _GFileBaseTest directly.")
-
- def testWith(self):
- with self.gfile(self.tmp + "test_with", "w") as fh:
- fh.write("hi")
- with self.gfile(self.tmp + "test_with", "r") as fh:
- self.assertEqual(fh.read(), "hi")
-
- def testSizeAndTellAndSeek(self):
- with self.gfile(self.tmp + "test_tell", "w") as fh:
- fh.write("".join(["0"] * 1000))
- with self.gfile(self.tmp + "test_tell", "r") as fh:
- self.assertEqual(1000, fh.Size())
- self.assertEqual(0, fh.tell())
- fh.seek(0, 2)
- self.assertEqual(1000, fh.tell())
- fh.seek(0)
- self.assertEqual(0, fh.tell())
-
- def testReadAndWritelines(self):
- with self.gfile(self.tmp + "test_writelines", "w") as fh:
- fh.writelines(["%d\n" % d for d in range(10)])
- with self.gfile(self.tmp + "test_writelines", "r") as fh:
- self.assertEqual(["%d\n" % x for x in range(10)], fh.readlines())
-
- def testWriteAndTruncate(self):
- with self.gfile(self.tmp + "test_truncate", "w") as fh:
- fh.write("ababab")
- with self.gfile(self.tmp + "test_truncate", "a+") as fh:
- fh.seek(0, 2)
- fh.write("hjhjhj")
- with self.gfile(self.tmp + "test_truncate", "a+") as fh:
- self.assertEqual(fh.Size(), 12)
- fh.truncate(6)
- with self.gfile(self.tmp + "test_truncate", "r") as fh:
- self.assertEqual(fh.read(), "ababab")
-
- def testErrors(self):
- self.assertRaises(
- IOError, lambda: self.gfile(self.tmp + "doesnt_exist", "r"))
- with self.gfile(self.tmp + "test_error", "w") as fh:
- # Raises FileError inside Google and ValueError outside, so we
- # can only test for Exception.
- self.assertRaises(Exception, lambda: fh.seek(-1))
- # test_error now exists, we can read from it:
- with self.gfile(self.tmp + "test_error", "r") as fh:
- self.assertRaises(IOError, lambda: fh.write("ack"))
- fh = self.gfile(self.tmp + "test_error", "w")
- self.assertFalse(fh.closed)
- fh.close()
- self.assertTrue(fh.closed)
- self.assertRaises(ValueError, lambda: fh.write("ack"))
-
- def testIteration(self):
- with self.gfile(self.tmp + "test_iter", "w") as fh:
- fh.writelines(["a\n", "b\n", "c\n"])
- with self.gfile(self.tmp + "test_iter", "r") as fh:
- lines = list(fh)
- self.assertEqual(["a\n", "b\n", "c\n"], lines)
-
-
-class GFileTest(_GFileBaseTest, googletest.TestCase):
-
- @property
- def gfile(self):
- return gfile.GFile
-
-
-class FastGFileTest(_GFileBaseTest, googletest.TestCase):
-
- @property
- def gfile(self):
- return gfile.FastGFile
-
-
-class FunctionTests(_BaseTest, googletest.TestCase):
-
- def testExists(self):
- self.assertFalse(gfile.Exists(self.tmp + "test_exists"))
- with gfile.GFile(self.tmp + "test_exists", "w"):
- pass
- self.assertTrue(gfile.Exists(self.tmp + "test_exists"))
-
- def testMkDirsGlobAndRmDirs(self):
- self.assertFalse(gfile.Exists(self.tmp + "test_dir"))
- gfile.MkDir(self.tmp + "test_dir")
- self.assertTrue(gfile.Exists(self.tmp + "test_dir"))
- gfile.RmDir(self.tmp + "test_dir")
- self.assertFalse(gfile.Exists(self.tmp + "test_dir"))
- gfile.MakeDirs(self.tmp + "test_dir/blah0")
- gfile.MakeDirs(self.tmp + "test_dir/blah1")
- self.assertEqual([self.tmp + "test_dir/blah0", self.tmp + "test_dir/blah1"],
- sorted(gfile.Glob(self.tmp + "test_dir/*")))
- gfile.DeleteRecursively(self.tmp + "test_dir")
- self.assertFalse(gfile.Exists(self.tmp + "test_dir"))
-
- @contextlib.contextmanager
- def _working_directory(self, wd):
- original_cwd = os.getcwd()
- os.chdir(wd)
- try:
- yield
- finally:
- os.chdir(original_cwd)
-
- def testMakeDirsWithEmptyString(self):
- gfile.MakeDirs(self.tmp + "test_dir")
- with self._working_directory(self.tmp + "test_dir"):
- gfile.MakeDirs("")
- # Should succeed because MakeDirs("") is a no-op.
- gfile.RmDir(self.tmp + "test_dir")
-
- def testErrors(self):
- self.assertRaises(
- OSError, lambda: gfile.RmDir(self.tmp + "dir_doesnt_exist"))
- self.assertRaises(
- OSError, lambda: gfile.Remove(self.tmp + "file_doesnt_exist"))
- gfile.MkDir(self.tmp + "error_dir")
- with gfile.GFile(self.tmp + "error_dir/file", "w"):
- pass # Create file
- self.assertRaises(
- OSError, lambda: gfile.Remove(self.tmp + "error_dir"))
- self.assertRaises(
- OSError, lambda: gfile.RmDir(self.tmp + "error_dir"))
- self.assertTrue(gfile.Exists(self.tmp + "error_dir"))
- gfile.DeleteRecursively(self.tmp + "error_dir")
- self.assertFalse(gfile.Exists(self.tmp + "error_dir"))
-
- def testStat(self):
- with gfile.GFile(self.tmp + "test_stat", "w"):
- pass
- creation_time = time.time()
- statinfo = gfile.Stat(self.tmp + "test_stat")
- # Test the modification timestamp is within 20 seconds of closing the file.
- self.assertLessEqual(statinfo.mtime, creation_time + 10)
- self.assertGreaterEqual(statinfo.mtime, creation_time - 10)
-
- def testRename(self):
- gfile.MkDir(self.tmp + "dir1")
- gfile.MkDir(self.tmp + "dir2")
- with gfile.GFile(self.tmp + "file1", "w"):
- pass # Create file
- with gfile.GFile(self.tmp + "file2", "w"):
- pass # Create file
-
- # Dest file already exists, overwrite=False (default).
- self.assertRaises(
- OSError, lambda: gfile.Rename(self.tmp + "file1", self.tmp + "file2"))
- gfile.Rename(self.tmp + "file1", self.tmp + "file2", overwrite=True)
- self.assertFalse(gfile.Exists(self.tmp + "file1"))
- gfile.Rename(self.tmp + "file2", self.tmp + "newfile")
- self.assertTrue(gfile.Exists(self.tmp + "newfile"))
-
- gfile.Rename(self.tmp + "dir1", self.tmp + "dir2")
- self.assertFalse(gfile.Exists(self.tmp + "dir1"))
- gfile.Rename(self.tmp + "dir2", self.tmp + "newdir")
- self.assertTrue(gfile.Exists(self.tmp + "newdir"))
-
- def testCopy(self):
- gfile.MkDir(self.tmp + "dir1")
- gfile.MkDir(self.tmp + "dir2")
- with gfile.GFile(self.tmp + "dir1/file1", "w"):
- pass # Create file
- with gfile.GFile(self.tmp + "dir2/file2", "w"):
- pass # Create file
-
- # Dest file already exists, overwrite=False (default).
- self.assertRaises(
- OSError, lambda: gfile.Copy(self.tmp + "dir1/file1",
- self.tmp + "dir2/file2"))
- # Overwrite succeeds
- gfile.Copy(self.tmp + "dir1/file1", self.tmp + "dir2/file2",
- overwrite=True)
- self.assertTrue(gfile.Exists(self.tmp + "dir2/file2"))
-
- # Normal copy.
- gfile.Rename(self.tmp + "dir1/file1", self.tmp + "dir2/file1")
- self.assertTrue(gfile.Exists(self.tmp + "dir2/file1"))
-
- # Normal copy to non-existent dir
- self.assertRaises(OSError,
- lambda: gfile.Rename(self.tmp + "dir1/file1",
- self.tmp + "newdir/file1"))
-
- def testOpen(self):
- with gfile.Open(self.tmp + "test_open", "wb") as f:
- f.write(b"foo")
- with gfile.Open(self.tmp + "test_open") as f:
- result = f.readlines()
- self.assertEqual(["foo"], result)
-
-if __name__ == "__main__":
- googletest.main()
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-26 18:10:35 +0000