diff options
Diffstat (limited to 'ufo/test/unit/common/test_fs_utils.py')
-rw-r--r-- | ufo/test/unit/common/test_fs_utils.py | 277 |
1 files changed, 277 insertions, 0 deletions
diff --git a/ufo/test/unit/common/test_fs_utils.py b/ufo/test/unit/common/test_fs_utils.py new file mode 100644 index 00000000000..186e07d59b6 --- /dev/null +++ b/ufo/test/unit/common/test_fs_utils.py @@ -0,0 +1,277 @@ +# Copyright (c) 2012 Red Hat, Inc. +# +# 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. + +import os +import shutil +import random +import unittest +from tempfile import mkdtemp, mkstemp +from gluster.swift.common import fs_utils as fs +from gluster.swift.common.exceptions import NotDirectoryError, \ + FileOrDirNotFoundError + +class TestUtils(unittest.TestCase): + """ Tests for common.utils """ + + def test_do_walk(self): + try: + # create directory structure + tmpparent = mkdtemp() + tmpdirs = [] + tmpfiles = [] + for i in range(5): + tmpdirs.append(mkdtemp(dir=tmpparent).rsplit(os.path.sep, 1)[1]) + tmpfiles.append(mkstemp(dir=tmpparent)[1].rsplit(os.path.sep, \ + 1)[1]) + + for path, dirnames, filenames in fs.do_walk(tmpparent): + assert path == tmpparent + assert dirnames.sort() == tmpdirs.sort() + assert filenames.sort() == tmpfiles.sort() + break + finally: + shutil.rmtree(tmpparent) + + def test_do_open(self): + try: + fd, tmpfile = mkstemp() + f = fs.do_open(tmpfile, 'r') + try: + f.write('test') + except IOError as err: + pass + else: + self.fail("IOError expected") + finally: + f.close() + os.close(fd) + os.remove(tmpfile) + + def test_do_open_err(self): + try: + fs.do_open(os.path.join('/tmp', str(random.random())), 'r') + except IOError: + pass + else: + self.fail("IOError expected") + + def test_do_write(self): + try: + fd, tmpfile = mkstemp() + cnt = fs.do_write(fd, "test") + assert cnt == len("test") + finally: + os.close(fd) + os.remove(tmpfile) + + def test_do_write_err(self): + try: + fd, tmpfile = mkstemp() + fd1 = os.open(tmpfile, os.O_RDONLY) + fs.do_write(fd1, "test") + except OSError: + pass + else: + self.fail("OSError expected") + finally: + os.close(fd) + os.close(fd1) + + def test_do_mkdir(self): + try: + path = os.path.join('/tmp', str(random.random())) + fs.do_mkdir(path) + assert os.path.exists(path) + assert fs.do_mkdir(path) + finally: + os.rmdir(path) + + def test_do_mkdir_err(self): + try: + path = os.path.join('/tmp', str(random.random()), str(random.random())) + fs.do_mkdir(path) + except OSError: + pass + else: + self.fail("OSError expected") + + + def test_do_makedirs(self): + try: + subdir = os.path.join('/tmp', str(random.random())) + path = os.path.join(subdir, str(random.random())) + fs.do_makedirs(path) + assert os.path.exists(path) + assert fs.do_makedirs(path) + finally: + shutil.rmtree(subdir) + + def test_do_listdir(self): + try: + tmpdir = mkdtemp() + subdir = [] + for i in range(5): + subdir.append(mkdtemp(dir=tmpdir).rsplit(os.path.sep, 1)[1]) + + assert subdir.sort() == fs.do_listdir(tmpdir).sort() + finally: + shutil.rmtree(tmpdir) + + def test_do_listdir_err(self): + try: + path = os.path.join('/tmp', str(random.random())) + fs.do_listdir(path) + except OSError: + pass + else: + self.fail("OSError expected") + + def test_do_stat(self): + try: + tmpdir = mkdtemp() + fd, tmpfile = mkstemp(dir=tmpdir) + buf1 = os.stat(tmpfile) + buf2 = fs.do_stat(fd) + buf3 = fs.do_stat(tmpfile) + + assert buf1 == buf2 + assert buf1 == buf3 + finally: + os.close(fd) + os.remove(tmpfile) + os.rmdir(tmpdir) + + def test_do_stat_err(self): + try: + fs.do_stat(os.path.join('/tmp', str(random.random()))) + except OSError: + pass + else: + self.fail("OSError expected") + + def test_do_close(self): + try: + fd, tmpfile = mkstemp() + fs.do_close(fd); + try: + os.write(fd, "test") + except OSError: + pass + else: + self.fail("OSError expected") + fp = open(tmpfile) + fs.do_close(fp) + finally: + os.remove(tmpfile) + + def test_do_unlink(self): + try: + fd, tmpfile = mkstemp() + fs.do_unlink(tmpfile) + assert not os.path.exists(tmpfile) + assert fs.do_unlink(os.path.join('/tmp', str(random.random()))) + finally: + os.close(fd) + + def test_do_unlink_err(self): + try: + tmpdir = mkdtemp() + fs.do_unlink(tmpdir) + except OSError: + pass + else: + self.fail('OSError expected') + finally: + os.rmdir(tmpdir) + + def test_do_rmdir(self): + tmpdir = mkdtemp() + fs.do_rmdir(tmpdir) + assert not os.path.exists(tmpdir) + assert not fs.do_rmdir(os.path.join('/tmp', str(random.random()))) + + def test_do_rmdir_err(self): + try: + fd, tmpfile = mkstemp() + fs.do_rmdir(tmpfile) + except OSError: + pass + else: + self.fail('OSError expected') + finally: + os.close(fd) + os.remove(tmpfile) + + def test_do_rename(self): + try: + srcpath = mkdtemp() + destpath = os.path.join('/tmp', str(random.random())) + fs.do_rename(srcpath, destpath) + assert not os.path.exists(srcpath) + assert os.path.exists(destpath) + finally: + os.rmdir(destpath) + + def test_do_rename_err(self): + try: + srcpath = os.path.join('/tmp', str(random.random())) + destpath = os.path.join('/tmp', str(random.random())) + fs.do_rename(srcpath, destpath) + except OSError: + pass + else: + self.fail("OSError expected") + + def test_dir_empty(self): + try: + tmpdir = mkdtemp() + subdir = mkdtemp(dir=tmpdir) + assert not fs.dir_empty(tmpdir) + assert fs.dir_empty(subdir) + finally: + shutil.rmtree(tmpdir) + + def test_dir_empty_err(self): + try: + try: + assert fs.dir_empty(os.path.join('/tmp', str(random.random()))) + except FileOrDirNotFoundError: + pass + else: + self.fail("FileOrDirNotFoundError exception expected") + + fd, tmpfile = mkstemp() + try: + fs.dir_empty(tmpfile) + except NotDirectoryError: + pass + else: + self.fail("NotDirectoryError exception expected") + finally: + os.close(fd) + os.unlink(tmpfile) + + def test_rmdirs(self): + try: + tmpdir = mkdtemp() + subdir = mkdtemp(dir=tmpdir) + fd, tmpfile = mkstemp(dir=tmpdir) + assert not fs.rmdirs(tmpfile) + assert not fs.rmdirs(tmpdir) + assert fs.rmdirs(subdir) + assert not os.path.exists(subdir) + finally: + os.close(fd) + shutil.rmtree(tmpdir) |