diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d912ada..9bd0a99 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -31,7 +31,7 @@ jobs: - name: Run tests with coverage run: | - coverage run -m unittest discover -s tests + coverage run -m unittest discover coverage xml # Generates the XML report for Codacy - name: Upload coverage to Codacy diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..e9c85a5 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,343 @@ +import errno +import os +import shutil +import stat +import tempfile +import unittest +import uuid + +from fuse import FuseOSError +from tempfile import mkdtemp +from unittest.mock import patch, MagicMock + +from globaleaks_eph_fs import EphemeralFile, EphemeralOperations, mount_globaleaks_eph_fs, main, unmount_if_mounted + +TEST_PATH = str(uuid.uuid4()) +TEST_DATA = b"Hello, world! This is a test data for writing, seeking and reading operations." + +ORIGINAL_SIZE = len(TEST_DATA) +EXTENDED_SIZE = ORIGINAL_SIZE*2 +REDUCED_SIZE = ORIGINAL_SIZE//2 + +class TestEphemeralFile(unittest.TestCase): + def setUp(self): + self.storage_dir = mkdtemp() + self.ephemeral_file = EphemeralFile(self.storage_dir) + + def tearDown(self): + shutil.rmtree(self.storage_dir) + + def test_create_and_write_file(self): + with self.ephemeral_file.open('w') as file: + file.write(TEST_DATA) + + self.assertTrue(os.path.exists(self.ephemeral_file.filepath)) + + def test_encryption_and_decryption(self): + with self.ephemeral_file.open('w') as file: + file.write(TEST_DATA) + + # Define test cases: each case is a tuple (seek_position, read_size, expected_data) + seek_tests = [ + (0, 1, TEST_DATA[:1]), # Seek at the start read 1 byte + (5, 5, TEST_DATA[5:10]), # Seek forward, read 5 bytes + (10, 2, TEST_DATA[10:12]), # Seek forward, read 2 bytes + (0, 3, TEST_DATA[:3]), # Seek backward, read 3 bytes + ] + + # Test forward and backward seeking with different offsets + with self.ephemeral_file.open('r') as file: + for seek_pos, read_size, expected in seek_tests: + file.seek(seek_pos) # Seek to the given position + self.assertEqual(file.tell(), seek_pos) # Check position after seeking forward + read_data = file.read(read_size) # Read the specified number of bytes + self.assertEqual(read_data, expected) # Verify the data matches the expected value + + def test_file_cleanup(self): + path_copy = self.ephemeral_file.filepath + del self.ephemeral_file + self.assertFalse(os.path.exists(path_copy)) + +class TestEphemeralOperations(unittest.TestCase): + def setUp(self): + self.storage_dir = mkdtemp() + 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.fs.files.values(): + os.remove(file.filepath) + os.rmdir(self.storage_dir) + + def test_create_file(self): + self.fs.create(TEST_PATH, 0o660) + self.assertIn(TEST_PATH, self.fs.files) + + def test_create_file_with_arbitrary_name(self): + with self.assertRaises(FuseOSError) as context: + self.fs.create('/arbitraryname', os.O_RDONLY) + self.assertEqual(context.exception.errno, errno.ENOENT) + + def test_open_existing_file(self): + self.fs.create(TEST_PATH, 0o660) + self.fs.open(TEST_PATH, os.O_RDONLY) + + def test_write_and_read_file(self): + self.fs.create(TEST_PATH, 0o660) + + self.fs.open(TEST_PATH, os.O_RDWR) + self.fs.write(TEST_PATH, TEST_DATA, 0, None) + + self.fs.release(TEST_PATH, None) + + self.fs.open(TEST_PATH, os.O_RDONLY) + + read_data = self.fs.read(TEST_PATH, len(TEST_DATA), 0, None) + + self.assertEqual(read_data, TEST_DATA) + + self.fs.release(TEST_PATH, None) + + def test_unlink_file(self): + self.fs.create(TEST_PATH, 0o660) + self.assertIn(TEST_PATH, self.fs.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.fs.open('/nonexistentfile', os.O_RDONLY) + self.assertEqual(context.exception.errno, errno.ENOENT) + + def test_getattr_root(self): + attr = self.fs.getattr('/') + self.assertEqual(stat.S_IFMT(attr['st_mode']), stat.S_IFDIR) + self.assertEqual(attr['st_mode'] & 0o777, 0o750) + self.assertEqual(attr['st_nlink'], 2) + + def test_getattr_file(self): + self.fs.create(TEST_PATH, mode=0o660) + + attr = self.fs.getattr(TEST_PATH) + + self.assertEqual(stat.S_IFMT(attr['st_mode']), stat.S_IFREG) + self.assertEqual(attr['st_mode'] & 0o777, 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.fs.getattr('/nonexistent') + + def test_truncate(self): + self.fs.create(TEST_PATH, 0o660) + self.fs.write(TEST_PATH, TEST_DATA, 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.fs.create(TEST_PATH, 0o660) + self.fs.write(TEST_PATH, TEST_DATA, 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 = [] + for _ in range(3): + file_names.append(str(uuid.uuid4())) + self.fs.create(file_names[-1], 0o660) + + directory_contents = self.fs.readdir('/', None) + self.assertEqual(set(directory_contents), {'.', '..', file_names[0], file_names[1], file_names[2]}) + + self.fs.unlink(file_names[1]) + directory_contents = self.fs.readdir('/', None) + self.assertEqual(set(directory_contents), {'.', '..', file_names[0], file_names[2]}) + + @patch("os.chmod") + def test_chmod_success(self, mock_chmod): + self.fs.create(TEST_PATH, 0o660) + mock_chmod.assert_called_with(self.fs.files[TEST_PATH].filepath, 0o660) + self.fs.chmod(TEST_PATH, 0o640) + mock_chmod.assert_called_with(self.fs.files[TEST_PATH].filepath, 0o640) + + 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.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) + + @patch('atexit.register') + @patch('argparse.ArgumentParser.parse_args') + @patch('globaleaks_eph_fs.subprocess.run') + @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') + @patch('globaleaks_eph_fs.FUSE') + @patch('builtins.print') + def test_main_mount_with_unspecified_storage_directory(self, mock_print, mock_FUSE, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): + with tempfile.TemporaryDirectory() as mount_point: + mock_parse_args.return_value = MagicMock( + mount_point=mount_point, + storage_directory=None, + unmount=False + ) + + original_mount_function = mount_globaleaks_eph_fs + + def side_effect_func(mount_point, storage_directory, flag): + return original_mount_function(mount_point, storage_directory, False) + + mock_mount.side_effect = side_effect_func + + main() + + mock_mount.assert_called_once_with(mount_point, None, True) + + mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) + + mock_subprocess.assert_called_once_with(['mount'], capture_output=True, text=True) + + @patch('atexit.register') + @patch('argparse.ArgumentParser.parse_args') + @patch('globaleaks_eph_fs.subprocess.run') + @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') + @patch('globaleaks_eph_fs.FUSE') + @patch('builtins.print') + def test_main_mount_with_specified_storage_directory(self, mock_print, mock_FUSE, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): + with tempfile.TemporaryDirectory() as mount_point, tempfile.TemporaryDirectory() as storage_directory: + mock_parse_args.return_value = MagicMock( + mount_point=mount_point, + storage_directory=storage_directory + ) + + original_mount_function = mount_globaleaks_eph_fs + + def side_effect_func(mount_point, storage_directory, flag): + return original_mount_function(mount_point, storage_directory, False) + + mock_mount.side_effect = side_effect_func + + main() + + mock_mount.assert_called_once_with(mount_point, storage_directory, True) + + mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) + + mock_subprocess.assert_called_once_with(['mount'], capture_output=True, text=True) + + @patch('atexit.register') + @patch('argparse.ArgumentParser.parse_args') + @patch('globaleaks_eph_fs.subprocess.run') + @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') + @patch('globaleaks_eph_fs.is_mount_point') + @patch('globaleaks_eph_fs.FUSE') + @patch('builtins.print') + def test_main_with_mount_point_check(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): + with tempfile.TemporaryDirectory() as mount_point: + mock_parse_args.return_value = MagicMock( + mount_point=mount_point, + storage_directory=None + ) + + mock_is_mount_point.return_value = True + + original_mount_function = mount_globaleaks_eph_fs + + def side_effect_func(mount_point, storage_directory, flag): + return original_mount_function(mount_point, storage_directory, False) + + mock_mount.side_effect = side_effect_func + + main() + + mock_mount.assert_called_once_with(mount_point, None, True) + + mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) + + mock_subprocess.assert_called_once_with(['fusermount', '-u', mount_point]) + + @patch('atexit.register') + @patch('argparse.ArgumentParser.parse_args') + @patch('globaleaks_eph_fs.subprocess.run') + @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') + @patch('globaleaks_eph_fs.is_mount_point') + @patch('globaleaks_eph_fs.FUSE') + @patch('builtins.print') + def test_main_keyboard_interrupt(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): + with tempfile.TemporaryDirectory() as mount_point: + mock_parse_args.return_value = MagicMock( + mount_point=mount_point, + storage_directory=None + ) + + mock_is_mount_point.return_value = False + + mock_mount.side_effect = KeyboardInterrupt + + with self.assertRaises(SystemExit): + main() + + mock_mount.assert_called_once_with(mount_point, None, True) + + mock_subprocess.assert_not_called() + + @patch('atexit.register') + @patch('argparse.ArgumentParser.parse_args') + @patch('globaleaks_eph_fs.subprocess.run') + @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') + @patch('globaleaks_eph_fs.is_mount_point') + @patch('globaleaks_eph_fs.FUSE') + @patch('builtins.print') + def test_main_other_exception(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): + with tempfile.TemporaryDirectory() as mount_point: + mock_parse_args.return_value = MagicMock( + mount_point=mount_point, + storage_directory=None + ) + + mock_is_mount_point.return_value = False + + mock_mount.side_effect = Exception("Some unexpected error") + + with self.assertRaises(SystemExit): + main() + + mock_mount.assert_called_once_with(mount_point, None, True) + + mock_atexit_register.assert_not_called() + + mock_subprocess.assert_not_called() + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test.py b/tests/test.py deleted file mode 100644 index e9c85a5..0000000 --- a/tests/test.py +++ /dev/null @@ -1,343 +0,0 @@ -import errno -import os -import shutil -import stat -import tempfile -import unittest -import uuid - -from fuse import FuseOSError -from tempfile import mkdtemp -from unittest.mock import patch, MagicMock - -from globaleaks_eph_fs import EphemeralFile, EphemeralOperations, mount_globaleaks_eph_fs, main, unmount_if_mounted - -TEST_PATH = str(uuid.uuid4()) -TEST_DATA = b"Hello, world! This is a test data for writing, seeking and reading operations." - -ORIGINAL_SIZE = len(TEST_DATA) -EXTENDED_SIZE = ORIGINAL_SIZE*2 -REDUCED_SIZE = ORIGINAL_SIZE//2 - -class TestEphemeralFile(unittest.TestCase): - def setUp(self): - self.storage_dir = mkdtemp() - self.ephemeral_file = EphemeralFile(self.storage_dir) - - def tearDown(self): - shutil.rmtree(self.storage_dir) - - def test_create_and_write_file(self): - with self.ephemeral_file.open('w') as file: - file.write(TEST_DATA) - - self.assertTrue(os.path.exists(self.ephemeral_file.filepath)) - - def test_encryption_and_decryption(self): - with self.ephemeral_file.open('w') as file: - file.write(TEST_DATA) - - # Define test cases: each case is a tuple (seek_position, read_size, expected_data) - seek_tests = [ - (0, 1, TEST_DATA[:1]), # Seek at the start read 1 byte - (5, 5, TEST_DATA[5:10]), # Seek forward, read 5 bytes - (10, 2, TEST_DATA[10:12]), # Seek forward, read 2 bytes - (0, 3, TEST_DATA[:3]), # Seek backward, read 3 bytes - ] - - # Test forward and backward seeking with different offsets - with self.ephemeral_file.open('r') as file: - for seek_pos, read_size, expected in seek_tests: - file.seek(seek_pos) # Seek to the given position - self.assertEqual(file.tell(), seek_pos) # Check position after seeking forward - read_data = file.read(read_size) # Read the specified number of bytes - self.assertEqual(read_data, expected) # Verify the data matches the expected value - - def test_file_cleanup(self): - path_copy = self.ephemeral_file.filepath - del self.ephemeral_file - self.assertFalse(os.path.exists(path_copy)) - -class TestEphemeralOperations(unittest.TestCase): - def setUp(self): - self.storage_dir = mkdtemp() - 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.fs.files.values(): - os.remove(file.filepath) - os.rmdir(self.storage_dir) - - def test_create_file(self): - self.fs.create(TEST_PATH, 0o660) - self.assertIn(TEST_PATH, self.fs.files) - - def test_create_file_with_arbitrary_name(self): - with self.assertRaises(FuseOSError) as context: - self.fs.create('/arbitraryname', os.O_RDONLY) - self.assertEqual(context.exception.errno, errno.ENOENT) - - def test_open_existing_file(self): - self.fs.create(TEST_PATH, 0o660) - self.fs.open(TEST_PATH, os.O_RDONLY) - - def test_write_and_read_file(self): - self.fs.create(TEST_PATH, 0o660) - - self.fs.open(TEST_PATH, os.O_RDWR) - self.fs.write(TEST_PATH, TEST_DATA, 0, None) - - self.fs.release(TEST_PATH, None) - - self.fs.open(TEST_PATH, os.O_RDONLY) - - read_data = self.fs.read(TEST_PATH, len(TEST_DATA), 0, None) - - self.assertEqual(read_data, TEST_DATA) - - self.fs.release(TEST_PATH, None) - - def test_unlink_file(self): - self.fs.create(TEST_PATH, 0o660) - self.assertIn(TEST_PATH, self.fs.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.fs.open('/nonexistentfile', os.O_RDONLY) - self.assertEqual(context.exception.errno, errno.ENOENT) - - def test_getattr_root(self): - attr = self.fs.getattr('/') - self.assertEqual(stat.S_IFMT(attr['st_mode']), stat.S_IFDIR) - self.assertEqual(attr['st_mode'] & 0o777, 0o750) - self.assertEqual(attr['st_nlink'], 2) - - def test_getattr_file(self): - self.fs.create(TEST_PATH, mode=0o660) - - attr = self.fs.getattr(TEST_PATH) - - self.assertEqual(stat.S_IFMT(attr['st_mode']), stat.S_IFREG) - self.assertEqual(attr['st_mode'] & 0o777, 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.fs.getattr('/nonexistent') - - def test_truncate(self): - self.fs.create(TEST_PATH, 0o660) - self.fs.write(TEST_PATH, TEST_DATA, 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.fs.create(TEST_PATH, 0o660) - self.fs.write(TEST_PATH, TEST_DATA, 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 = [] - for _ in range(3): - file_names.append(str(uuid.uuid4())) - self.fs.create(file_names[-1], 0o660) - - directory_contents = self.fs.readdir('/', None) - self.assertEqual(set(directory_contents), {'.', '..', file_names[0], file_names[1], file_names[2]}) - - self.fs.unlink(file_names[1]) - directory_contents = self.fs.readdir('/', None) - self.assertEqual(set(directory_contents), {'.', '..', file_names[0], file_names[2]}) - - @patch("os.chmod") - def test_chmod_success(self, mock_chmod): - self.fs.create(TEST_PATH, 0o660) - mock_chmod.assert_called_with(self.fs.files[TEST_PATH].filepath, 0o660) - self.fs.chmod(TEST_PATH, 0o640) - mock_chmod.assert_called_with(self.fs.files[TEST_PATH].filepath, 0o640) - - 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.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) - - @patch('atexit.register') - @patch('argparse.ArgumentParser.parse_args') - @patch('globaleaks_eph_fs.subprocess.run') - @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') - @patch('globaleaks_eph_fs.FUSE') - @patch('builtins.print') - def test_main_mount_with_unspecified_storage_directory(self, mock_print, mock_FUSE, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): - with tempfile.TemporaryDirectory() as mount_point: - mock_parse_args.return_value = MagicMock( - mount_point=mount_point, - storage_directory=None, - unmount=False - ) - - original_mount_function = mount_globaleaks_eph_fs - - def side_effect_func(mount_point, storage_directory, flag): - return original_mount_function(mount_point, storage_directory, False) - - mock_mount.side_effect = side_effect_func - - main() - - mock_mount.assert_called_once_with(mount_point, None, True) - - mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) - - mock_subprocess.assert_called_once_with(['mount'], capture_output=True, text=True) - - @patch('atexit.register') - @patch('argparse.ArgumentParser.parse_args') - @patch('globaleaks_eph_fs.subprocess.run') - @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') - @patch('globaleaks_eph_fs.FUSE') - @patch('builtins.print') - def test_main_mount_with_specified_storage_directory(self, mock_print, mock_FUSE, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): - with tempfile.TemporaryDirectory() as mount_point, tempfile.TemporaryDirectory() as storage_directory: - mock_parse_args.return_value = MagicMock( - mount_point=mount_point, - storage_directory=storage_directory - ) - - original_mount_function = mount_globaleaks_eph_fs - - def side_effect_func(mount_point, storage_directory, flag): - return original_mount_function(mount_point, storage_directory, False) - - mock_mount.side_effect = side_effect_func - - main() - - mock_mount.assert_called_once_with(mount_point, storage_directory, True) - - mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) - - mock_subprocess.assert_called_once_with(['mount'], capture_output=True, text=True) - - @patch('atexit.register') - @patch('argparse.ArgumentParser.parse_args') - @patch('globaleaks_eph_fs.subprocess.run') - @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') - @patch('globaleaks_eph_fs.is_mount_point') - @patch('globaleaks_eph_fs.FUSE') - @patch('builtins.print') - def test_main_with_mount_point_check(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): - with tempfile.TemporaryDirectory() as mount_point: - mock_parse_args.return_value = MagicMock( - mount_point=mount_point, - storage_directory=None - ) - - mock_is_mount_point.return_value = True - - original_mount_function = mount_globaleaks_eph_fs - - def side_effect_func(mount_point, storage_directory, flag): - return original_mount_function(mount_point, storage_directory, False) - - mock_mount.side_effect = side_effect_func - - main() - - mock_mount.assert_called_once_with(mount_point, None, True) - - mock_atexit_register.assert_called_once_with(unmount_if_mounted, mount_point) - - mock_subprocess.assert_called_once_with(['fusermount', '-u', mount_point]) - - @patch('atexit.register') - @patch('argparse.ArgumentParser.parse_args') - @patch('globaleaks_eph_fs.subprocess.run') - @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') - @patch('globaleaks_eph_fs.is_mount_point') - @patch('globaleaks_eph_fs.FUSE') - @patch('builtins.print') - def test_main_keyboard_interrupt(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): - with tempfile.TemporaryDirectory() as mount_point: - mock_parse_args.return_value = MagicMock( - mount_point=mount_point, - storage_directory=None - ) - - mock_is_mount_point.return_value = False - - mock_mount.side_effect = KeyboardInterrupt - - with self.assertRaises(SystemExit): - main() - - mock_mount.assert_called_once_with(mount_point, None, True) - - mock_subprocess.assert_not_called() - - @patch('atexit.register') - @patch('argparse.ArgumentParser.parse_args') - @patch('globaleaks_eph_fs.subprocess.run') - @patch('globaleaks_eph_fs.mount_globaleaks_eph_fs') - @patch('globaleaks_eph_fs.is_mount_point') - @patch('globaleaks_eph_fs.FUSE') - @patch('builtins.print') - def test_main_other_exception(self, mock_print, mock_FUSE, mock_is_mount_point, mock_mount, mock_subprocess, mock_parse_args, mock_atexit_register): - with tempfile.TemporaryDirectory() as mount_point: - mock_parse_args.return_value = MagicMock( - mount_point=mount_point, - storage_directory=None - ) - - mock_is_mount_point.return_value = False - - mock_mount.side_effect = Exception("Some unexpected error") - - with self.assertRaises(SystemExit): - main() - - mock_mount.assert_called_once_with(mount_point, None, True) - - mock_atexit_register.assert_not_called() - - mock_subprocess.assert_not_called() - -if __name__ == '__main__': - unittest.main()