Synnefo » snf-ganeti » ganeti-local

def TailFile(fname, lines=20):

  """Return the last lines from a file.

  @note: this function will only read and parse the last 4KB of

      the file; if the lines are very long, it could be that less

      than the requested number of lines are returned

  @param fname: the file name

  @type lines: int

  @param lines: the (maximum) number of lines to return

  """

  fd = open(fname, "r")

  try:

    fd.seek(0, 2)

    pos = fd.tell()

    pos = max(0, pos-4096)

    fd.seek(pos, 0)

    raw_data = fd.read()

  finally:

    fd.close()

  rows = raw_data.splitlines()

  return rows[-lines:]

     SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress, \

     TailFile

class TestTailFile(testutils.GanetiTestCase):

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

  def testEmpty(self):

    fname = self._CreateTempFile()

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

    self.failUnlessEqual(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(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(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(TailFile(fname, lines=i), data[-i:])