Implement chmod and chown commands

src/globaleaks_eph_fs/__init__.py

@@ -326,6 +326,32 @@ def truncate(self, path, length, fh=None):
                     original_file.write(b'\0' * to_write)
                     bytes_written += to_write
 
+    def chmod(self, path, mode):
+        """
+        Changes the permissions of the file at the specified path.
+
+        :param path: The file path whose permissions will be changed.
+        :param mode: The new permissions mode (e.g., 0o777 for full permissions).
+        :raises FuseOSError: If the file does not exist.
+        """
+        file = self.files.get(path)
+        if not file:
+            raise FuseOSError(errno.ENOENT)
+        return os.chmod(file.filepath, mode)
+
+    def chown(self, path, uid, gid):
+        """
+        Changes the ownership of the file at the specified path.
+
+        :param path: The file path whose ownership will be changed.
+        :param uid: The user ID (uid) to set as the new owner.
+        :param gid: The group ID (gid) to set as the new group owner.
+        :raises FuseOSError: If the file does not exist.
+        """
+        file = self.files.get(path)
+        if not file:
+            raise FuseOSError(errno.ENOENT)
+        return os.chown(file.filepath, uid, gid)
 
 class EphemeralFS(FUSE):
     """

tests/test.py

@@ -3,8 +3,11 @@
 import shutil
 import stat
 import unittest
-from tempfile import mkdtemp
+
 from fuse import FuseOSError
+from tempfile import mkdtemp
+from unittest.mock import patch, MagicMock
+
 from globaleaks_eph_fs import EphemeralFile, EphemeralOperations
 
 TEST_PATH = 'TESTFILE.TXT'
@@ -56,103 +59,140 @@ def test_file_cleanup(self):
 class TestEphemeralOperations(unittest.TestCase):
     def setUp(self):
         self.storage_dir = mkdtemp()
-        self.operations = EphemeralOperations(self.storage_dir)
+        self.fs = EphemeralOperations(self.storage_dir)
+
+        # Get current user's UID and GID
+        self.current_uid = os.getuid()
+        self.current_gid = os.getgid()
 
     def tearDown(self):
-        for file in self.operations.files.values():
+        for file in self.fs.files.values():
             os.remove(file.filepath)
         os.rmdir(self.storage_dir)
 
     def test_create_file(self):
-        self.operations.create(TEST_PATH, 0o660)
-        self.assertIn(TEST_PATH, self.operations.files)
+        self.fs.create(TEST_PATH, 0o660)
+        self.assertIn(TEST_PATH, self.fs.files)
 
     def test_open_existing_file(self):
-        self.operations.create(TEST_PATH, 0o660)
-        self.operations.open(TEST_PATH, os.O_RDONLY)
+        self.fs.create(TEST_PATH, 0o660)
+        self.fs.open(TEST_PATH, os.O_RDONLY)
 
     def test_write_and_read_file(self):
-        self.operations.create(TEST_PATH, 0o660)
+        self.fs.create(TEST_PATH, 0o660)
 
-        self.operations.open(TEST_PATH, os.O_RDWR)
-        self.operations.write(TEST_PATH, TEST_DATA, 0, None)
+        self.fs.open(TEST_PATH, os.O_RDWR)
+        self.fs.write(TEST_PATH, TEST_DATA, 0, None)
 
-        self.operations.release(TEST_PATH, None)
+        self.fs.release(TEST_PATH, None)
 
-        self.operations.open(TEST_PATH, os.O_RDONLY)
+        self.fs.open(TEST_PATH, os.O_RDONLY)
 
-        read_data = self.operations.read(TEST_PATH, len(TEST_DATA), 0, None)
+        read_data = self.fs.read(TEST_PATH, len(TEST_DATA), 0, None)
 
         self.assertEqual(read_data, TEST_DATA)
 
-        self.operations.release(TEST_PATH, None)
+        self.fs.release(TEST_PATH, None)
 
     def test_unlink_file(self):
-        self.operations.create(TEST_PATH, 0o660)
-        self.assertIn(TEST_PATH, self.operations.files)
+        self.fs.create(TEST_PATH, 0o660)
+        self.assertIn(TEST_PATH, self.fs.files)
 
-        self.operations.unlink(TEST_PATH)
-        self.assertNotIn(TEST_PATH, self.operations.files)
+        self.fs.unlink(TEST_PATH)
+        self.assertNotIn(TEST_PATH, self.fs.files)
 
     def test_file_not_found(self):
         with self.assertRaises(FuseOSError) as context:
-            self.operations.open('/nonexistentfile', os.O_RDONLY)
+            self.fs.open('/nonexistentfile', os.O_RDONLY)
         self.assertEqual(context.exception.errno, errno.ENOENT)
 
     def test_getattr_root(self):
-        attr = self.operations.getattr('/')
+        attr = self.fs.getattr('/')
         self.assertEqual(attr['st_mode'], stat.S_IFDIR | 0o750)
         self.assertEqual(attr['st_nlink'], 2)
 
     def test_getattr_file(self):
-        self.operations.create(TEST_PATH, mode=0o660)
+        self.fs.create(TEST_PATH, mode=0o660)
 
-        attr = self.operations.getattr(TEST_PATH)
+        attr = self.fs.getattr(TEST_PATH)
 
         self.assertEqual(attr['st_mode'], stat.S_IFREG | 0o660)
         self.assertEqual(attr['st_size'], 0)
         self.assertEqual(attr['st_nlink'], 1)
         self.assertEqual(attr['st_uid'], os.getuid())
         self.assertEqual(attr['st_gid'], os.getgid())
-
         self.assertIn('st_atime', attr)
         self.assertIn('st_mtime', attr)
         self.assertIn('st_ctime', attr)
 
     def test_getattr_nonexistent(self):
         with self.assertRaises(OSError) as _:
-            self.operations.getattr('/nonexistent')
+            self.fs.getattr('/nonexistent')
 
     def test_truncate(self):
-        self.operations.create(TEST_PATH, 0o660)
-        self.operations.write(TEST_PATH, TEST_DATA, 0, None)
+        self.fs.create(TEST_PATH, 0o660)
+        self.fs.write(TEST_PATH, TEST_DATA, 0, None)
 
-        self.operations.truncate(TEST_PATH, REDUCED_SIZE, None)
-        file_content = self.operations.read(TEST_PATH, ORIGINAL_SIZE, 0, None)
+        self.fs.truncate(TEST_PATH, REDUCED_SIZE, None)
+        file_content = self.fs.read(TEST_PATH, ORIGINAL_SIZE, 0, None)
         self.assertEqual(len(file_content), REDUCED_SIZE)
         self.assertEqual(file_content, TEST_DATA[:REDUCED_SIZE])
 
     def test_extend(self):
-        self.operations.create(TEST_PATH, 0o660)
-        self.operations.write(TEST_PATH, TEST_DATA, 0, None)
+        self.fs.create(TEST_PATH, 0o660)
+        self.fs.write(TEST_PATH, TEST_DATA, 0, None)
 
-        self.operations.truncate(TEST_PATH, EXTENDED_SIZE, None)
-        file_content = self.operations.read(TEST_PATH, EXTENDED_SIZE * 2, 0, None)
+        self.fs.truncate(TEST_PATH, EXTENDED_SIZE, None)
+        file_content = self.fs.read(TEST_PATH, EXTENDED_SIZE * 2, 0, None)
         self.assertEqual(file_content[:ORIGINAL_SIZE], TEST_DATA)
         self.assertEqual(len(file_content), EXTENDED_SIZE)
         self.assertTrue(all(byte == 0 for byte in file_content[ORIGINAL_SIZE:]))
 
     def test_readdir(self):
         file_names = ['/file1', '/file2', '/file3']
         for file_name in file_names:
-            self.operations.create(file_name, 0o660)
+            self.fs.create(file_name, 0o660)
 
-        directory_contents = self.operations.readdir('/', None)
+        directory_contents = self.fs.readdir('/', None)
         self.assertEqual(set(directory_contents), {'.', '..', 'file1', 'file2', 'file3'})
 
-        self.operations.unlink('/file2')
-        directory_contents = self.operations.readdir('/', None)
+        self.fs.unlink('/file2')
+        directory_contents = self.fs.readdir('/', None)
         self.assertEqual(set(directory_contents), {'.', '..', 'file1', 'file3'})
 
+    @patch("os.chmod")
+    def test_chmod_success(self, mock_chmod):
+        self.fs.create(TEST_PATH, 0o660)
+        self.fs.chmod(TEST_PATH, 0o644)
+        mock_chmod.assert_called_once_with(self.fs.files[TEST_PATH].filepath, 0o644)
+
+    def test_chmod_file_not_found(self):
+        with self.assertRaises(FuseOSError) as context:
+            self.fs.chmod("/nonexistent", 0o644)
+        self.assertEqual(context.exception.errno, errno.ENOENT)
+
+    @patch("os.chmod", side_effect=PermissionError)
+    def test_chmod_permission_error(self, mock_chmod):
+        self.fs.create(TEST_PATH, 0o660)
+        with self.assertRaises(PermissionError):
+            self.fs.chmod(TEST_PATH, 0o644)
+
+    @patch("os.chown")
+    def test_chown_success(self, mock_chown):
+        self.fs.create(TEST_PATH, 0o660)
+        self.fs.chown(TEST_PATH, self.current_uid, self.current_gid)
+        mock_chown.assert_called_once_with(self.fs.files[TEST_PATH].filepath, self.current_uid, self.current_gid)
+
+    def test_chown_file_not_found(self):
+        with self.assertRaises(FuseOSError) as context:
+            self.fs.chown("/nonexistent", self.current_uid, self.current_gid)
+        self.assertEqual(context.exception.errno, errno.ENOENT)
+
+    @patch("os.chown", side_effect=PermissionError)
+    def test_chown_permission_error(self, mock_chown):
+        self.fs.create(TEST_PATH, 0o660)
+        with self.assertRaises(PermissionError):
+            self.fs.chown(TEST_PATH, self.current_uid, self.current_gid)
+
 if __name__ == '__main__':
     unittest.main()