Add method last_bytes in fileutils

Method last_bytes is used in some projects[1]. It's good to bring
this method in fileutils according to discussion in dev ML[2].

[1] http://codesearch.openstack.org/?q=last_bytes&i=nope&files=&repos=
[2] http://lists.openstack.org/pipermail/openstack-dev/2017-July/120259.html

Change-Id: I1cd61de58b759916ecd0569afb2485de0b31c405

2 files changed, 50 insertions, 0 deletions

diff --git a/oslo_utils/fileutils.py b/oslo_utils/fileutils.py
index ed1bba9..3b0cdfc 100644
--- a/oslo_utils/fileutils.py
+++ b/oslo_utils/fileutils.py
@@ -124,3 +124,27 @@ def compute_file_checksum(path, read_chunksize=65536, algorithm='sha256'):
         for chunk in iter(lambda: f.read(read_chunksize), b''):
             checksum.update(chunk)
     return checksum.hexdigest()
+
+
+def last_bytes(path, num):
+    """Return num bytes from the end of the file, and unread byte count.
+
+    :param path: The file path to read
+    :param num: The number of bytes to return
+
+    :returns: (data, unread_bytes)
+    """
+
+    with open(path, 'rb') as fp:
+        try:
+            fp.seek(-num, os.SEEK_END)
+        except IOError as e:
+            # seek() fails with EINVAL when trying to go before the start of
+            # the file. It means that num is larger than the file size, so
+            # just go to the start.
+            if e.errno == errno.EINVAL:
+                fp.seek(0, os.SEEK_SET)
+            else:
+                raise
+        unread_bytes = fp.tell()
+        return (fp.read(), unread_bytes)
diff --git a/oslo_utils/tests/test_fileutils.py b/oslo_utils/tests/test_fileutils.py
index 33974ad..d34a38b 100644
--- a/oslo_utils/tests/test_fileutils.py
+++ b/oslo_utils/tests/test_fileutils.py
@@ -244,3 +244,29 @@ class TestComputeFileChecksum(test_base.BaseTestCase):
     def test_generic_io_error(self):
         tempdir = tempfile.mkdtemp()
         self.assertRaises(IOError, fileutils.compute_file_checksum, tempdir)
+
+
+class LastBytesTestCase(test_base.BaseTestCase):
+    """Test the last_bytes() utility method."""
+
+    def setUp(self):
+        super(LastBytesTestCase, self).setUp()
+        self.content = b'1234567890'
+
+    def test_truncated(self):
+        res = fileutils.write_to_tempfile(self.content)
+        self.assertTrue(os.path.exists(res))
+        out, unread_bytes = fileutils.last_bytes(res, 5)
+        self.assertEqual(b'67890', out)
+        self.assertGreater(unread_bytes, 0)
+
+    def test_read_all(self):
+        res = fileutils.write_to_tempfile(self.content)
+        self.assertTrue(os.path.exists(res))
+        out, unread_bytes = fileutils.last_bytes(res, 1000)
+        self.assertEqual(b'1234567890', out)
+        self.assertEqual(0, unread_bytes)
+
+    def test_non_exist_file(self):
+        self.assertRaises(IOError, fileutils.last_bytes,
+                          'non_exist_file', 1000)