Synnefo » snf-ganeti » ganeti-local

#!/usr/bin/python

#

# Copyright (C) 2006, 2007, 2010, 2011 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.utils.io"""

import os

import tempfile

import unittest

import shutil

import glob

import time

import signal

import stat

import errno

from ganeti import constants

from ganeti import utils

from ganeti import compat

from ganeti import errors

import testutils

class TestReadFile(testutils.GanetiTestCase):

  def testReadAll(self):

    data = utils.ReadFile(self._TestDataFilename("cert1.pem"))

    self.assertEqual(len(data), 814)

    h = compat.md5_hash()

    h.update(data)

    self.assertEqual(h.hexdigest(), "a491efb3efe56a0535f924d5f8680fd4")

  def testReadSize(self):

    data = utils.ReadFile(self._TestDataFilename("cert1.pem"),

                          size=100)

    self.assertEqual(len(data), 100)

    h = compat.md5_hash()

    h.update(data)

    self.assertEqual(h.hexdigest(), "893772354e4e690b9efd073eed433ce7")

  def testCallback(self):

    def _Cb(fh):

      self.assertEqual(fh.tell(), 0)

    data = utils.ReadFile(self._TestDataFilename("cert1.pem"), preread=_Cb)

    self.assertEqual(len(data), 814)

  def testError(self):

    self.assertRaises(EnvironmentError, utils.ReadFile,

                      "/dev/null/does-not-exist")

class TestReadOneLineFile(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

  def testDefault(self):

    data = utils.ReadOneLineFile(self._TestDataFilename("cert1.pem"))

    self.assertEqual(len(data), 27)

    self.assertEqual(data, "-----BEGIN CERTIFICATE-----")

  def testNotStrict(self):

    data = utils.ReadOneLineFile(self._TestDataFilename("cert1.pem"),

                                 strict=False)

    self.assertEqual(len(data), 27)

    self.assertEqual(data, "-----BEGIN CERTIFICATE-----")

  def testStrictFailure(self):

    self.assertRaises(errors.GenericError, utils.ReadOneLineFile,

                      self._TestDataFilename("cert1.pem"), strict=True)

  def testLongLine(self):

    dummydata = (1024 * "Hello World! ")

    myfile = self._CreateTempFile()

    utils.WriteFile(myfile, data=dummydata)

    datastrict = utils.ReadOneLineFile(myfile, strict=True)

    datalax = utils.ReadOneLineFile(myfile, strict=False)

    self.assertEqual(dummydata, datastrict)

    self.assertEqual(dummydata, datalax)

  def testNewline(self):

    myfile = self._CreateTempFile()

    myline = "myline"

    for nl in ["", "\n", "\r\n"]:

      dummydata = "%s%s" % (myline, nl)

      utils.WriteFile(myfile, data=dummydata)

      datalax = utils.ReadOneLineFile(myfile, strict=False)

      self.assertEqual(myline, datalax)

      datastrict = utils.ReadOneLineFile(myfile, strict=True)

      self.assertEqual(myline, datastrict)

  def testWhitespaceAndMultipleLines(self):

    myfile = self._CreateTempFile()

    for nl in ["", "\n", "\r\n"]:

      for ws in [" ", "\t", "\t\t  \t", "\t "]:

        dummydata = (1024 * ("Foo bar baz %s%s" % (ws, nl)))

        utils.WriteFile(myfile, data=dummydata)

        datalax = utils.ReadOneLineFile(myfile, strict=False)

        if nl:

          self.assert_(set("\r\n") & set(dummydata))

          self.assertRaises(errors.GenericError, utils.ReadOneLineFile,

                            myfile, strict=True)

          explen = len("Foo bar baz ") + len(ws)

          self.assertEqual(len(datalax), explen)

          self.assertEqual(datalax, dummydata[:explen])

          self.assertFalse(set("\r\n") & set(datalax))

        else:

          datastrict = utils.ReadOneLineFile(myfile, strict=True)

          self.assertEqual(dummydata, datastrict)

          self.assertEqual(dummydata, datalax)

  def testEmptylines(self):

    myfile = self._CreateTempFile()

    myline = "myline"

    for nl in ["\n", "\r\n"]:

      for ol in ["", "otherline"]:

        dummydata = "%s%s%s%s%s%s" % (nl, nl, myline, nl, ol, nl)

        utils.WriteFile(myfile, data=dummydata)

        self.assert_(set("\r\n") & set(dummydata))

        datalax = utils.ReadOneLineFile(myfile, strict=False)

        self.assertEqual(myline, datalax)

        if ol:

          self.assertRaises(errors.GenericError, utils.ReadOneLineFile,

                            myfile, strict=True)

        else:

          datastrict = utils.ReadOneLineFile(myfile, strict=True)

          self.assertEqual(myline, datastrict)

  def testEmptyfile(self):

    myfile = self._CreateTempFile()

    self.assertRaises(errors.GenericError, utils.ReadOneLineFile, myfile)

class TestTimestampForFilename(unittest.TestCase):

  def test(self):

    self.assert_("." not in utils.TimestampForFilename())

    self.assert_(":" not in utils.TimestampForFilename())

class TestCreateBackup(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    testutils.GanetiTestCase.tearDown(self)

    shutil.rmtree(self.tmpdir)

  def testEmpty(self):

    filename = utils.PathJoin(self.tmpdir, "config.data")

    utils.WriteFile(filename, data="")

    bname = utils.CreateBackup(filename)

    self.assertFileContent(bname, "")

    self.assertEqual(len(glob.glob("%s*" % filename)), 2)

    utils.CreateBackup(filename)

    self.assertEqual(len(glob.glob("%s*" % filename)), 3)

    utils.CreateBackup(filename)

    self.assertEqual(len(glob.glob("%s*" % filename)), 4)

    fifoname = utils.PathJoin(self.tmpdir, "fifo")

    os.mkfifo(fifoname)

    self.assertRaises(errors.ProgrammerError, utils.CreateBackup, fifoname)

  def testContent(self):

    bkpcount = 0

    for data in ["", "X", "Hello World!\n" * 100, "Binary data\0\x01\x02\n"]:

      for rep in [1, 2, 10, 127]:

        testdata = data * rep

        filename = utils.PathJoin(self.tmpdir, "test.data_")

        utils.WriteFile(filename, data=testdata)

        self.assertFileContent(filename, testdata)

        for _ in range(3):

          bname = utils.CreateBackup(filename)

          bkpcount += 1

          self.assertFileContent(bname, testdata)

          self.assertEqual(len(glob.glob("%s*" % filename)), 1 + bkpcount)

class TestListVisibleFiles(unittest.TestCase):

  """Test case for ListVisibleFiles"""

  def setUp(self):

    self.path = tempfile.mkdtemp()

  def tearDown(self):

    shutil.rmtree(self.path)

  def _CreateFiles(self, files):

    for name in files:

      utils.WriteFile(os.path.join(self.path, name), data="test")

  def _test(self, files, expected):

    self._CreateFiles(files)

    found = utils.ListVisibleFiles(self.path)

    self.assertEqual(set(found), set(expected))

  def testAllVisible(self):

    files = ["a", "b", "c"]

    expected = files

    self._test(files, expected)

  def testNoneVisible(self):

    files = [".a", ".b", ".c"]

    expected = []

    self._test(files, expected)

  def testSomeVisible(self):

    files = ["a", "b", ".c"]

    expected = ["a", "b"]

    self._test(files, expected)

  def testNonAbsolutePath(self):

    self.failUnlessRaises(errors.ProgrammerError, utils.ListVisibleFiles,

                          "abc")

  def testNonNormalizedPath(self):

    self.failUnlessRaises(errors.ProgrammerError, utils.ListVisibleFiles,

                          "/bin/../tmp")

  def testMountpoint(self):

    lvfmp_fn = compat.partial(utils.ListVisibleFiles,

                              _is_mountpoint=lambda _: True)

    self.assertEqual(lvfmp_fn(self.path), [])

    # Create "lost+found" as a regular file

    self._CreateFiles(["foo", "bar", ".baz", "lost+found"])

    self.assertEqual(set(lvfmp_fn(self.path)),

                     set(["foo", "bar", "lost+found"]))

    # Replace "lost+found" with a directory

    laf_path = utils.PathJoin(self.path, "lost+found")

    utils.RemoveFile(laf_path)

    os.mkdir(laf_path)

    self.assertEqual(set(lvfmp_fn(self.path)), set(["foo", "bar"]))

  def testLostAndFoundNoMountpoint(self):

    files = ["foo", "bar", ".Hello World", "lost+found"]

    expected = ["foo", "bar", "lost+found"]

    self._test(files, expected)

class TestWriteFile(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

    self.tmpdir = None

    self.tfile = tempfile.NamedTemporaryFile()

    self.did_pre = False

    self.did_post = False

    self.did_write = False

  def tearDown(self):

    testutils.GanetiTestCase.tearDown(self)

    if self.tmpdir:

      shutil.rmtree(self.tmpdir)

  def markPre(self, fd):

    self.did_pre = True

  def markPost(self, fd):

    self.did_post = True

  def markWrite(self, fd):

    self.did_write = True

  def testWrite(self):

    data = "abc"

    utils.WriteFile(self.tfile.name, data=data)

    self.assertEqual(utils.ReadFile(self.tfile.name), data)

  def testWriteSimpleUnicode(self):

    data = u"abc"

    utils.WriteFile(self.tfile.name, data=data)

    self.assertEqual(utils.ReadFile(self.tfile.name), data)

  def testErrors(self):

    self.assertRaises(errors.ProgrammerError, utils.WriteFile,

                      self.tfile.name, data="test", fn=lambda fd: None)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile, self.tfile.name)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile,

                      self.tfile.name, data="test", atime=0)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile, self.tfile.name,

                      mode=0400, keep_perms=utils.KP_ALWAYS)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile, self.tfile.name,

                      uid=0, keep_perms=utils.KP_ALWAYS)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile, self.tfile.name,

                      gid=0, keep_perms=utils.KP_ALWAYS)

    self.assertRaises(errors.ProgrammerError, utils.WriteFile, self.tfile.name,

                      mode=0400, uid=0, keep_perms=utils.KP_ALWAYS)

  def testPreWrite(self):

    utils.WriteFile(self.tfile.name, data="", prewrite=self.markPre)

    self.assertTrue(self.did_pre)

    self.assertFalse(self.did_post)

    self.assertFalse(self.did_write)

  def testPostWrite(self):

    utils.WriteFile(self.tfile.name, data="", postwrite=self.markPost)

    self.assertFalse(self.did_pre)

    self.assertTrue(self.did_post)

    self.assertFalse(self.did_write)

  def testWriteFunction(self):

    utils.WriteFile(self.tfile.name, fn=self.markWrite)

    self.assertFalse(self.did_pre)

    self.assertFalse(self.did_post)

    self.assertTrue(self.did_write)

  def testDryRun(self):

    orig = "abc"

    self.tfile.write(orig)

    self.tfile.flush()

    utils.WriteFile(self.tfile.name, data="hello", dry_run=True)

    self.assertEqual(utils.ReadFile(self.tfile.name), orig)

  def testTimes(self):

    f = self.tfile.name

    for at, mt in [(0, 0), (1000, 1000), (2000, 3000),

                   (int(time.time()), 5000)]:

      utils.WriteFile(f, data="hello", atime=at, mtime=mt)

      st = os.stat(f)

      self.assertEqual(st.st_atime, at)

      self.assertEqual(st.st_mtime, mt)

  def testNoClose(self):

    data = "hello"

    self.assertEqual(utils.WriteFile(self.tfile.name, data="abc"), None)

    fd = utils.WriteFile(self.tfile.name, data=data, close=False)

    try:

      os.lseek(fd, 0, 0)

      self.assertEqual(os.read(fd, 4096), data)

    finally:

      os.close(fd)

  def testNoLeftovers(self):

    self.tmpdir = tempfile.mkdtemp()

    self.assertEqual(utils.WriteFile(utils.PathJoin(self.tmpdir, "test"),

                                     data="abc"),

                     None)

    self.assertEqual(os.listdir(self.tmpdir), ["test"])

  def testFailRename(self):

    self.tmpdir = tempfile.mkdtemp()

    target = utils.PathJoin(self.tmpdir, "target")

    os.mkdir(target)

    self.assertRaises(OSError, utils.WriteFile, target, data="abc")

    self.assertTrue(os.path.isdir(target))

    self.assertEqual(os.listdir(self.tmpdir), ["target"])

    self.assertFalse(os.listdir(target))

  def testFailRenameDryRun(self):

    self.tmpdir = tempfile.mkdtemp()

    target = utils.PathJoin(self.tmpdir, "target")

    os.mkdir(target)

    self.assertEqual(utils.WriteFile(target, data="abc", dry_run=True), None)

    self.assertTrue(os.path.isdir(target))

    self.assertEqual(os.listdir(self.tmpdir), ["target"])

    self.assertFalse(os.listdir(target))

  def testBackup(self):

    self.tmpdir = tempfile.mkdtemp()

    testfile = utils.PathJoin(self.tmpdir, "test")

    self.assertEqual(utils.WriteFile(testfile, data="foo", backup=True), None)

    self.assertEqual(utils.ReadFile(testfile), "foo")

    self.assertEqual(os.listdir(self.tmpdir), ["test"])

    # Write again

    assert os.path.isfile(testfile)

    self.assertEqual(utils.WriteFile(testfile, data="bar", backup=True), None)

    self.assertEqual(utils.ReadFile(testfile), "bar")

    self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)

    self.assertTrue("test" in os.listdir(self.tmpdir))

    self.assertEqual(len(os.listdir(self.tmpdir)), 2)

    # Write again as dry-run

    assert os.path.isfile(testfile)

    self.assertEqual(utils.WriteFile(testfile, data="000", backup=True,

                                     dry_run=True),

                     None)

    self.assertEqual(utils.ReadFile(testfile), "bar")

    self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)

    self.assertTrue("test" in os.listdir(self.tmpdir))

    self.assertEqual(len(os.listdir(self.tmpdir)), 2)

  def testFileMode(self):

    self.tmpdir = tempfile.mkdtemp()

    target = utils.PathJoin(self.tmpdir, "target")

    self.assertRaises(OSError, utils.WriteFile, target, data="data",

                      keep_perms=utils.KP_ALWAYS)

    # All masks have only user bits set, to avoid interactions with umask

    utils.WriteFile(target, data="data", mode=0200)

    self.assertFileMode(target, 0200)

    utils.WriteFile(target, data="data", mode=0400,

                    keep_perms=utils.KP_IF_EXISTS)

    self.assertFileMode(target, 0200)

    utils.WriteFile(target, data="data", keep_perms=utils.KP_ALWAYS)

    self.assertFileMode(target, 0200)

    utils.WriteFile(target, data="data", mode=0700)

    self.assertFileMode(target, 0700)

  def testNewFileMode(self):

    self.tmpdir = tempfile.mkdtemp()

    target = utils.PathJoin(self.tmpdir, "target")

    utils.WriteFile(target, data="data", mode=0400,

                    keep_perms=utils.KP_IF_EXISTS)

    self.assertFileMode(target, 0400)

class TestFileID(testutils.GanetiTestCase):

  def testEquality(self):

    name = self._CreateTempFile()

    oldi = utils.GetFileID(path=name)

    self.failUnless(utils.VerifyFileID(oldi, oldi))

  def testUpdate(self):

    name = self._CreateTempFile()

    oldi = utils.GetFileID(path=name)

    fd = os.open(name, os.O_RDWR)

    try:

      newi = utils.GetFileID(fd=fd)

      self.failUnless(utils.VerifyFileID(oldi, newi))

      self.failUnless(utils.VerifyFileID(newi, oldi))

    finally:

      os.close(fd)

  def testWriteFile(self):

    name = self._CreateTempFile()

    oldi = utils.GetFileID(path=name)

    mtime = oldi[2]

    os.utime(name, (mtime + 10, mtime + 10))

    self.assertRaises(errors.LockError, utils.SafeWriteFile, name,

                      oldi, data="")

    os.utime(name, (mtime - 10, mtime - 10))

    utils.SafeWriteFile(name, oldi, data="")

    oldi = utils.GetFileID(path=name)

    mtime = oldi[2]

    os.utime(name, (mtime + 10, mtime + 10))

    # this doesn't raise, since we passed None

    utils.SafeWriteFile(name, None, data="")

  def testError(self):

    t = tempfile.NamedTemporaryFile()

    self.assertRaises(errors.ProgrammerError, utils.GetFileID,

                      path=t.name, fd=t.fileno())

class TestRemoveFile(unittest.TestCase):

  """Test case for the RemoveFile function"""

  def setUp(self):

    """Create a temp dir and file for each case"""

    self.tmpdir = tempfile.mkdtemp('', 'ganeti-unittest-')

    fd, self.tmpfile = tempfile.mkstemp('', '', self.tmpdir)

    os.close(fd)

  def tearDown(self):

    if os.path.exists(self.tmpfile):

      os.unlink(self.tmpfile)

    os.rmdir(self.tmpdir)

  def testIgnoreDirs(self):

    """Test that RemoveFile() ignores directories"""

    self.assertEqual(None, utils.RemoveFile(self.tmpdir))

  def testIgnoreNotExisting(self):

    """Test that RemoveFile() ignores non-existing files"""

    utils.RemoveFile(self.tmpfile)

    utils.RemoveFile(self.tmpfile)

  def testRemoveFile(self):

    """Test that RemoveFile does remove a file"""

    utils.RemoveFile(self.tmpfile)

    if os.path.exists(self.tmpfile):

      self.fail("File '%s' not removed" % self.tmpfile)

  def testRemoveSymlink(self):

    """Test that RemoveFile does remove symlinks"""

    symlink = self.tmpdir + "/symlink"

    os.symlink("no-such-file", symlink)

    utils.RemoveFile(symlink)

    if os.path.exists(symlink):

      self.fail("File '%s' not removed" % symlink)

    os.symlink(self.tmpfile, symlink)

    utils.RemoveFile(symlink)

    if os.path.exists(symlink):

      self.fail("File '%s' not removed" % symlink)

class TestRemoveDir(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    try:

      shutil.rmtree(self.tmpdir)

    except EnvironmentError:

      pass

  def testEmptyDir(self):

    utils.RemoveDir(self.tmpdir)

    self.assertFalse(os.path.isdir(self.tmpdir))

  def testNonEmptyDir(self):

    self.tmpfile = os.path.join(self.tmpdir, "test1")

    open(self.tmpfile, "w").close()

    self.assertRaises(EnvironmentError, utils.RemoveDir, self.tmpdir)

class TestRename(unittest.TestCase):

  """Test case for RenameFile"""

  def setUp(self):

    """Create a temporary directory"""

    self.tmpdir = tempfile.mkdtemp()

    self.tmpfile = os.path.join(self.tmpdir, "test1")

    # Touch the file

    open(self.tmpfile, "w").close()

  def tearDown(self):

    """Remove temporary directory"""

    shutil.rmtree(self.tmpdir)

  def testSimpleRename1(self):

    """Simple rename 1"""

    utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"))

    self.assert_(os.path.isfile(os.path.join(self.tmpdir, "xyz")))

  def testSimpleRename2(self):

    """Simple rename 2"""

    utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"),

                     mkdir=True)

    self.assert_(os.path.isfile(os.path.join(self.tmpdir, "xyz")))

  def testRenameMkdir(self):

    """Rename with mkdir"""

    utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "test/xyz"),

                     mkdir=True)

    self.assert_(os.path.isdir(os.path.join(self.tmpdir, "test")))

    self.assert_(os.path.isfile(os.path.join(self.tmpdir, "test/xyz")))

    self.assertRaises(EnvironmentError, utils.RenameFile,

                      os.path.join(self.tmpdir, "test/xyz"),

                      os.path.join(self.tmpdir, "test/foo/bar/baz"),

                      mkdir=True)

    self.assertTrue(os.path.exists(os.path.join(self.tmpdir, "test/xyz")))

    self.assertFalse(os.path.exists(os.path.join(self.tmpdir, "test/foo/bar")))

    self.assertFalse(os.path.exists(os.path.join(self.tmpdir,

                                                 "test/foo/bar/baz")))

class TestMakedirs(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def testNonExisting(self):

    path = utils.PathJoin(self.tmpdir, "foo")

    utils.Makedirs(path)

    self.assert_(os.path.isdir(path))

  def testExisting(self):

    path = utils.PathJoin(self.tmpdir, "foo")

    os.mkdir(path)

    utils.Makedirs(path)

    self.assert_(os.path.isdir(path))

  def testRecursiveNonExisting(self):

    path = utils.PathJoin(self.tmpdir, "foo/bar/baz")

    utils.Makedirs(path)

    self.assert_(os.path.isdir(path))

  def testRecursiveExisting(self):

    path = utils.PathJoin(self.tmpdir, "B/moo/xyz")

    self.assertFalse(os.path.exists(path))

    os.mkdir(utils.PathJoin(self.tmpdir, "B"))

    utils.Makedirs(path)

    self.assert_(os.path.isdir(path))

class TestEnsureDirs(unittest.TestCase):

  """Tests for EnsureDirs"""

  def setUp(self):

    self.dir = tempfile.mkdtemp()

    self.old_umask = os.umask(0777)

  def testEnsureDirs(self):

    utils.EnsureDirs([

        (utils.PathJoin(self.dir, "foo"), 0777),

        (utils.PathJoin(self.dir, "bar"), 0000),

        ])

    self.assertEquals(os.stat(utils.PathJoin(self.dir, "foo"))[0] & 0777, 0777)

    self.assertEquals(os.stat(utils.PathJoin(self.dir, "bar"))[0] & 0777, 0000)

  def tearDown(self):

    os.rmdir(utils.PathJoin(self.dir, "foo"))

    os.rmdir(utils.PathJoin(self.dir, "bar"))

    os.rmdir(self.dir)

    os.umask(self.old_umask)

class TestIsNormAbsPath(unittest.TestCase):

  """Testing case for IsNormAbsPath"""

  def _pathTestHelper(self, path, result):

    if result:

      self.assert_(utils.IsNormAbsPath(path),

          "Path %s should result absolute and normalized" % path)

    else:

      self.assertFalse(utils.IsNormAbsPath(path),

          "Path %s should not result absolute and normalized" % path)

  def testBase(self):

    self._pathTestHelper("/etc", True)

    self._pathTestHelper("/srv", True)

    self._pathTestHelper("etc", False)

    self._pathTestHelper("/etc/../root", False)

    self._pathTestHelper("/etc/", False)

class TestIsBelowDir(unittest.TestCase):

  """Testing case for IsBelowDir"""

  def testExactlyTheSame(self):

    self.assertFalse(utils.IsBelowDir("/a/b", "/a/b"))

    self.assertFalse(utils.IsBelowDir("/a/b", "/a/b/"))

    self.assertFalse(utils.IsBelowDir("/a/b/", "/a/b"))

    self.assertFalse(utils.IsBelowDir("/a/b/", "/a/b/"))

  def testSamePrefix(self):

    self.assertTrue(utils.IsBelowDir("/a/b", "/a/b/c"))

    self.assertTrue(utils.IsBelowDir("/a/b/", "/a/b/e"))

  def testSamePrefixButDifferentDir(self):

    self.assertFalse(utils.IsBelowDir("/a/b", "/a/bc/d"))

    self.assertFalse(utils.IsBelowDir("/a/b/", "/a/bc/e"))

  def testSamePrefixButDirTraversal(self):

    self.assertFalse(utils.IsBelowDir("/a/b", "/a/b/../c"))

    self.assertFalse(utils.IsBelowDir("/a/b/", "/a/b/../d"))

  def testSamePrefixAndTraversal(self):

    self.assertTrue(utils.IsBelowDir("/a/b", "/a/b/c/../d"))

    self.assertTrue(utils.IsBelowDir("/a/b", "/a/b/c/./e"))

    self.assertTrue(utils.IsBelowDir("/a/b", "/a/b/../b/./e"))

  def testBothAbsPath(self):

    self.assertRaises(ValueError, utils.IsBelowDir, "/a/b/c", "d")

    self.assertRaises(ValueError, utils.IsBelowDir, "a/b/c", "/d")

    self.assertRaises(ValueError, utils.IsBelowDir, "a/b/c", "d")

class TestPathJoin(unittest.TestCase):

  """Testing case for PathJoin"""

  def testBasicItems(self):

    mlist = ["/a", "b", "c"]

    self.failUnlessEqual(utils.PathJoin(*mlist), "/".join(mlist))

  def testNonAbsPrefix(self):

    self.failUnlessRaises(ValueError, utils.PathJoin, "a", "b")

  def testBackTrack(self):

    self.failUnlessRaises(ValueError, utils.PathJoin, "/a", "b/../c")

  def testMultiAbs(self):

    self.failUnlessRaises(ValueError, utils.PathJoin, "/a", "/b")

class TestTailFile(testutils.GanetiTestCase):

  """Test case for the TailFile function"""

  def testEmpty(self):

    fname = self._CreateTempFile()

    self.failUnlessEqual(utils.TailFile(fname), [])

    self.failUnlessEqual(utils.TailFile(fname, lines=25), [])

  def testAllLines(self):

    data = ["test %d" % i for i in range(30)]

    for i in range(30):

      fname = self._CreateTempFile()

      fd = open(fname, "w")

      fd.write("\n".join(data[:i]))

      if i > 0:

        fd.write("\n")

      fd.close()

      self.failUnlessEqual(utils.TailFile(fname, lines=i), data[:i])

  def testPartialLines(self):

    data = ["test %d" % i for i in range(30)]

    fname = self._CreateTempFile()

    fd = open(fname, "w")

    fd.write("\n".join(data))

    fd.write("\n")

    fd.close()

    for i in range(1, 30):

      self.failUnlessEqual(utils.TailFile(fname, lines=i), data[-i:])

  def testBigFile(self):

    data = ["test %d" % i for i in range(30)]

    fname = self._CreateTempFile()

    fd = open(fname, "w")

    fd.write("X" * 1048576)

    fd.write("\n")

    fd.write("\n".join(data))

    fd.write("\n")

    fd.close()

    for i in range(1, 30):

      self.failUnlessEqual(utils.TailFile(fname, lines=i), data[-i:])