Adds basic unit tests for:

 1. `read_file`
 2. `read_file_lines`
 3. `get_file_size`
 4. `delete_file`
 5. `read_head_lines`
 6. `read_tail_lines`
 7. `make_symlink`
reusing the existing structure.

Includes a TODO to fix get_file_size to actually return result value on
exception as it currently doesn't and therefore effectively returns None.

Author: jonasbardino

tests/test_mig_shared_fileio.py

@@ -46,6 +46,18 @@
 DUMMY_UNICODE_LENGTH = len(DUMMY_UNICODE)
 DUMMY_FILE_WRITECHUNK = 'fileio/write_chunk'
 DUMMY_FILE_WRITEFILE = 'fileio/write_file'
+# TODO: add similar tests for write_file_lines and enable next
+# DUMMY_FILE_WRITEFILELINES = 'fileio/write_file_lines'
+DUMMY_FILE_READFILE = 'fileio/read_file'
+DUMMY_FILE_READFILELINES = 'fileio/read_file_lines'
+DUMMY_FILE_READHEADLINES = 'fileio/read_head_lines'
+DUMMY_FILE_READTAILLINES = 'fileio/read_tail_lines'
+DUMMY_FILE_DELETEFILE = 'fileio/delete_file'
+DUMMY_FILE_GETFILESIZE = 'fileio/get_file_size'
+DUMMY_FILE_MAKESYMLINKSRC = 'fileio/make_symlink/link'
+DUMMY_FILE_MAKESYMLINKDST = 'fileio/make_symlink/target'
+# NOTE: getsize returns 4k for directories
+DUMMY_DIRECTORY_SIZE = 4096
 
 assert isinstance(DUMMY_BYTES, bytes)
 
@@ -57,6 +69,8 @@ def setUp(self):
         """Initialize test environment for write_chunk tests"""
         super(MigSharedFileio__write_chunk, self).setUp()
         self.tmp_path = temppath(DUMMY_FILE_WRITECHUNK, self)
+        # Output dir is created by default here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
         cleanpath(os.path.dirname(DUMMY_FILE_WRITECHUNK), self)
 
     def test_return_false_on_invalid_data(self):
@@ -155,6 +169,7 @@ def setUp(self):
         """Initialize test environment for write_file tests"""
         super(MigSharedFileio__write_file, self).setUp()
         self.tmp_path = temppath(DUMMY_FILE_WRITEFILE, self)
+        # Output dir is created by default here
         cleanpath(os.path.dirname(DUMMY_FILE_WRITEFILE), self)
 
     def test_return_false_on_invalid_data(self):
@@ -246,5 +261,273 @@ def test_store_unicode_in_binary_mode(self):
             self.assertEqual(content[:], DUMMY_UNICODE)
 
 
+class MigSharedFileio__read_file(MigTestCase):
+    """Test the read_file function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for read_file tests"""
+        super(MigSharedFileio__read_file, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_READFILE, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_path), self)
+
+    def test_reads_bytes(self):
+        """Test read_file returns byte content with binary mode"""
+        with open(self.tmp_path, 'wb') as fh:
+            fh.write(DUMMY_BYTES)
+        content = fileio.read_file(self.tmp_path, self.logger, mode='rb')
+        self.assertEqual(content, DUMMY_BYTES)
+
+    def test_reads_text(self):
+        """Test read_file returns text with text mode"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write(DUMMY_UNICODE)
+        content = fileio.read_file(self.tmp_path, self.logger, mode='r')
+        self.assertEqual(content, DUMMY_UNICODE)
+
+    def test_allows_missing_file(self):
+        """Test read_file returns None with allow_missing=True"""
+        content = fileio.read_file(
+            'missing.txt', self.logger, allow_missing=True)
+        self.assertIsNone(content)
+
+    def test_reports_missing_file(self):
+        """Test read_file returns None with allow_missing=False"""
+        self.logger.forgive_errors()
+        content = fileio.read_file(
+            'missing.txt', self.logger, allow_missing=False)
+        self.assertIsNone(content)
+
+    def test_handles_directory_path(self):
+        """Test read_file returns None when path is directory"""
+        self.logger.forgive_errors()
+        os.makedirs(self.tmp_path)
+        content = fileio.read_file(self.tmp_path, self.logger)
+        self.assertIsNone(content)
+
+
+class MigSharedFileio__read_file_lines(MigTestCase):
+    """Test the read_file_lines function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for read_file_lines tests"""
+        super(MigSharedFileio__read_file_lines, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_READFILELINES, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_path), self)
+
+    def test_returns_empty_list_for_empty_file(self):
+        """Test read_file_lines returns empty list for empty file"""
+        open(self.tmp_path, 'w').close()
+        lines = fileio.read_file_lines(self.tmp_path, self.logger)
+        self.assertEqual(lines, [])
+
+    def test_reads_lines_from_file(self):
+        """Test read_file_lines returns lines from text file"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write("line1\nline2\nline3")
+        lines = fileio.read_file_lines(self.tmp_path, self.logger)
+        self.assertEqual(lines, ["line1\n", "line2\n", "line3"])
+
+    def test_none_for_missing_file(self):
+        self.logger.forgive_errors()
+        lines = fileio.read_file_lines('missing.txt', self.logger)
+        self.assertIsNone(lines)
+
+
+class MigSharedFileio__get_file_size(MigTestCase):
+    """Test the get_file_size function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for get_file_size tests"""
+        super(MigSharedFileio__get_file_size, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_GETFILESIZE, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_path), self)
+
+    def test_returns_file_size(self):
+        """Test get_file_size returns correct file size"""
+        with open(self.tmp_path, 'wb') as fh:
+            fh.write(DUMMY_BYTES)
+        size = fileio.get_file_size(self.tmp_path, self.logger)
+        self.assertEqual(size, DUMMY_BYTES_LENGTH)
+
+    def test_handles_missing_file(self):
+        """Test get_file_size returns -1 for missing file"""
+        self.logger.forgive_errors()
+        size = fileio.get_file_size('missing.txt', self.logger)
+        # TODO: fix called function to return on exception and enable next line
+        # self.assertEqual(size, -1)
+        self.assertIsNone(size)
+
+    def test_handles_directory(self):
+        """Test get_file_size returns -1 when path is directory"""
+        self.logger.forgive_errors()
+        os.makedirs(self.tmp_path)
+        size = fileio.get_file_size(self.tmp_path, self.logger)
+        self.assertEqual(size, DUMMY_DIRECTORY_SIZE)
+
+
+class MigSharedFileio__delete_file(MigTestCase):
+    """Test the delete_file function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for delete_file tests"""
+        super(MigSharedFileio__delete_file, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_DELETEFILE, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(DUMMY_FILE_DELETEFILE), self)
+
+    def test_deletes_existing_file(self):
+        """Test delete_file removes existing file"""
+        open(self.tmp_path, 'w').close()
+        result = fileio.delete_file(self.tmp_path, self.logger)
+        self.assertTrue(result)
+        self.assertFalse(os.path.exists(self.tmp_path))
+
+    def test_handles_missing_file_with_allow_missing(self):
+        """Test delete_file succeeds with allow_missing=True"""
+        result = fileio.delete_file(
+            'missing.txt', self.logger, allow_missing=True)
+        self.assertTrue(result)
+
+    def test_false_for_missing_file_without_allow_missing(self):
+        """Test delete_file returns False with allow_missing=False"""
+        self.logger.forgive_errors()
+        result = fileio.delete_file('missing.txt',
+                                    self.logger,
+                                    allow_missing=False)
+        self.assertFalse(result)
+
+
+class MigSharedFileio__read_head_lines(MigTestCase):
+    """Test the read_head_lines function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for read_head_lines tests"""
+        super(MigSharedFileio__read_head_lines, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_READHEADLINES, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_path), self)
+
+    def test_reads_requested_lines(self):
+        """Test read_head_lines returns requested number of lines"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write("line1\nline2\nline3\nline4")
+        lines = fileio.read_head_lines(self.tmp_path, 2, self.logger)
+        self.assertEqual(lines, ["line1\n", "line2\n"])
+
+    def test_returns_all_lines_when_requested_more(self):
+        """Test read_head_lines returns all lines when file has fewer"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write("line1\nline2")
+        lines = fileio.read_head_lines(self.tmp_path, 5, self.logger)
+        self.assertEqual(lines, ["line1\n", "line2"])
+
+    def test_returns_empty_list_for_empty_file(self):
+        """Test read_head_lines returns empty for empty file"""
+        open(self.tmp_path, 'w').close()
+        lines = fileio.read_head_lines(self.tmp_path, 3, self.logger)
+        self.assertEqual(lines, [])
+
+    def test_empty_for_missing_file(self):
+        """Test read_head_lines returns [] for missing file"""
+        self.logger.forgive_errors()
+        lines = fileio.read_head_lines('missing.txt', 3, self.logger)
+        self.assertEqual(lines, [])
+
+
+class MigSharedFileio__read_tail_lines(MigTestCase):
+    """Test the read_tail_lines function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for read_tail_lines tests"""
+        super(MigSharedFileio__read_tail_lines, self).setUp()
+        self.tmp_path = temppath(DUMMY_FILE_READTAILLINES, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_path), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_path), self)
+
+    def test_reads_requested_lines(self):
+        """Test read_tail_lines returns requested number of lines"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write("line1\nline2\nline3\nline4")
+        lines = fileio.read_tail_lines(self.tmp_path, 2, self.logger)
+        self.assertEqual(lines, ["line3\n", "line4"])
+
+    def test_returns_all_lines_when_requested_more(self):
+        """Test read_tail_lines returns all lines when file has fewer"""
+        with open(self.tmp_path, 'w') as fh:
+            fh.write("line1\nline2")
+        lines = fileio.read_tail_lines(self.tmp_path, 5, self.logger)
+        self.assertEqual(lines, ["line1\n", "line2"])
+
+    def test_returns_empty_list_for_empty_file(self):
+        """Test read_tail_lines returns empty for empty file"""
+        open(self.tmp_path, 'w').close()
+        lines = fileio.read_tail_lines(self.tmp_path, 3, self.logger)
+        self.assertEqual(lines, [])
+
+    def test_empty_for_missing_file(self):
+        """Test read_tail_lines returns [] for missing file"""
+        self.logger.forgive_errors()
+        lines = fileio.read_tail_lines('missing.txt', 3, self.logger)
+        self.assertEqual(lines, [])
+
+
+class MigSharedFileio__make_symlink(MigTestCase):
+    """Test the make_symlink function from mig.shared.fileio module"""
+
+    def setUp(self):
+        """Initialize test environment for make_symlink tests"""
+        super(MigSharedFileio__make_symlink, self).setUp()
+        self.tmp_link = temppath(DUMMY_FILE_MAKESYMLINKSRC, self)
+        self.tmp_target = temppath(DUMMY_FILE_MAKESYMLINKDST, self)
+        # We generally need output dir to exist here
+        os.makedirs(os.path.dirname(self.tmp_target), exist_ok=True)
+        cleanpath(os.path.dirname(self.tmp_link), self)
+        cleanpath(os.path.dirname(self.tmp_target), self)
+        with open(self.tmp_target, 'w') as fh:
+            fh.write("test")
+
+    def test_creates_symlink(self):
+        """Test make_symlink creates working symlink"""
+        result = fileio.make_symlink(
+            self.tmp_target, self.tmp_link, self.logger)
+        self.assertTrue(result)
+        self.assertTrue(os.path.islink(self.tmp_link))
+        self.assertEqual(os.readlink(self.tmp_link), self.tmp_target)
+
+    def test_force_overwrites_existing_link(self):
+        """Test make_symlink force replaces existing link"""
+        os.symlink('/dummy', self.tmp_link)
+        result = fileio.make_symlink(self.tmp_target, self.tmp_link, self.logger,
+                                     force=True)
+        self.assertTrue(result)
+        self.assertEqual(os.readlink(self.tmp_link), self.tmp_target)
+
+    def test_fails_on_existing_link_without_force(self):
+        """Test make_symlink fails on existing link without force"""
+        self.logger.forgive_errors()
+        os.symlink('/dummy', self.tmp_link)
+        result = fileio.make_symlink(self.tmp_target, self.tmp_link, self.logger,
+                                     force=False)
+        self.assertFalse(result)
+
+    def test_handles_nonexistent_target(self):
+        """Test make_symlink still creates broken symlink"""
+        self.logger.forgive_errors()
+        broken_target = self.tmp_target + '-nonexistent'
+        result = fileio.make_symlink(broken_target, self.tmp_link, self.logger)
+        self.assertTrue(result)
+        self.assertTrue(os.path.islink(self.tmp_link))
+        self.assertEqual(os.readlink(self.tmp_link), broken_target)
+
+
 if __name__ == '__main__':
     testmain()