#!/usr/bin/env python3 """ Additional tests for FileManager covering uncovered paths: - _safe_path traversal rejection - create_user: duplicate (file already has user), invalid username - delete_user: not in file, invalid username - list_users with actual htpasswd file - get_users from users.json - _get_user_storage_info for nonexistent dir - _list_user_folders - create_folder / delete_folder edge cases - upload_file / download_file / delete_file edge cases - list_files edge cases - backup_user_files invalid username - restore_user_files invalid username / nonexistent backup - get_status in docker and non-docker mode - _test_filesystem_access - _test_user_authentication - test_connectivity - get_webdav_status (mocked subprocess) """ import sys import os import shutil import tempfile import unittest from pathlib import Path from unittest.mock import patch, MagicMock api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) from file_manager import FileManager class TestFileManagerExtra(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.data_dir = os.path.join(self.tmp, 'data') self.config_dir = os.path.join(self.tmp, 'config') os.makedirs(self.data_dir, exist_ok=True) os.makedirs(self.config_dir, exist_ok=True) self.fm = FileManager(data_dir=self.data_dir, config_dir=self.config_dir) def tearDown(self): shutil.rmtree(self.tmp) # ── _safe_path ────────────────────────────────────────────────────────── def test_safe_path_traversal_raises(self): with self.assertRaises(ValueError): self.fm._safe_path('alice', '../../../etc/passwd') def test_safe_path_invalid_username_raises(self): with self.assertRaises(ValueError): self.fm._safe_path('user with space', 'file.txt') def test_safe_path_valid_returns_path_under_files_dir(self): path = self.fm._safe_path('alice', 'Documents', 'readme.txt') self.assertTrue(path.startswith(self.fm.files_dir)) # ── create_user ───────────────────────────────────────────────────────── def test_create_user_invalid_username_chars_returns_false(self): result = self.fm.create_user('bad user!', 'password') self.assertFalse(result) def test_create_user_creates_default_folders(self): result = self.fm.create_user('alice', 'secret') self.assertTrue(result) user_dir = os.path.join(self.fm.files_dir, 'alice') for folder in ('Documents', 'Pictures', 'Music', 'Videos', 'Downloads'): self.assertTrue(os.path.isdir(os.path.join(user_dir, folder))) def test_create_user_bcrypt_hash_uses_2y_prefix(self): self.fm.create_user('alice', 'secret') auth_file = os.path.join(self.fm.webdav_dir, 'users') with open(auth_file) as f: line = f.read() self.assertIn('$2y$', line) self.assertNotIn('$2b$', line) def test_create_user_appends_to_existing_file(self): self.fm.create_user('alice', 'secret') self.fm.create_user('bob', 'secret2') auth_file = os.path.join(self.fm.webdav_dir, 'users') with open(auth_file) as f: lines = [l for l in f if l.strip()] self.assertEqual(len(lines), 2) # ── delete_user ───────────────────────────────────────────────────────── def test_delete_user_invalid_username_returns_false(self): result = self.fm.delete_user('bad user!') self.assertFalse(result) def test_delete_user_not_in_auth_file_still_returns_true(self): """Delete succeeds even if user was never in the auth file.""" result = self.fm.delete_user('nobody') self.assertTrue(result) def test_delete_user_removes_only_matching_line(self): self.fm.create_user('alice', 'secret') self.fm.create_user('bob', 'secret2') self.fm.delete_user('alice') auth_file = os.path.join(self.fm.webdav_dir, 'users') with open(auth_file) as f: content = f.read() self.assertNotIn('alice:', content) self.assertIn('bob:', content) # ── list_users ────────────────────────────────────────────────────────── def test_list_users_returns_usernames_from_auth_file(self): self.fm.create_user('alice', 'secret') self.fm.create_user('bob', 'secret2') users = self.fm.list_users() usernames = [u['username'] for u in users] self.assertIn('alice', usernames) self.assertIn('bob', usernames) def test_list_users_skips_malformed_lines(self): auth_file = os.path.join(self.fm.webdav_dir, 'users') with open(auth_file, 'w') as f: f.write('alicehash\n') # no colon f.write('bob:hash\n') users = self.fm.list_users() usernames = [u['username'] for u in users] self.assertNotIn('alicehash', usernames) self.assertIn('bob', usernames) # ── get_users ─────────────────────────────────────────────────────────── def test_get_users_returns_empty_when_no_file(self): result = self.fm.get_users() self.assertEqual(result, []) def test_get_users_returns_list_from_json(self): import json webdav_dir = os.path.join(self.config_dir, 'webdav') os.makedirs(webdav_dir, exist_ok=True) users_file = os.path.join(webdav_dir, 'users.json') with open(users_file, 'w') as f: json.dump([{'username': 'alice'}], f) result = self.fm.get_users() self.assertEqual(len(result), 1) self.assertEqual(result[0]['username'], 'alice') # ── _get_user_storage_info ─────────────────────────────────────────────── def test_storage_info_nonexistent_dir_returns_zeros(self): info = self.fm._get_user_storage_info('nobody') self.assertEqual(info['total_files'], 0) self.assertEqual(info['total_size_bytes'], 0) def test_storage_info_counts_files(self): self.fm.create_user('alice', 'secret') self.fm.upload_file('alice', 'Documents/a.txt', b'hello') self.fm.upload_file('alice', 'Documents/b.txt', b'world') info = self.fm._get_user_storage_info('alice') self.assertGreaterEqual(info['total_files'], 2) # ── _list_user_folders ─────────────────────────────────────────────────── def test_list_user_folders_returns_list(self): self.fm.create_user('alice', 'secret') folders = self.fm._list_user_folders('alice') self.assertIsInstance(folders, list) names = [f['name'] for f in folders] self.assertIn('Documents', names) def test_list_user_folders_empty_for_nonexistent_user(self): folders = self.fm._list_user_folders('nobody') self.assertEqual(folders, []) # ── create_folder / delete_folder edge cases ──────────────────────────── def test_create_folder_traversal_returns_false(self): self.fm.create_user('alice', 'secret') result = self.fm.create_folder('alice', '../../../tmp/evil') self.assertFalse(result) def test_delete_folder_nonexistent_returns_false(self): self.fm.create_user('alice', 'secret') result = self.fm.delete_folder('alice', 'NoSuchFolder') self.assertFalse(result) def test_delete_folder_traversal_returns_false(self): self.fm.create_user('alice', 'secret') result = self.fm.delete_folder('alice', '../../../tmp/evil') self.assertFalse(result) # ── upload / download / delete edge cases ─────────────────────────────── def test_download_file_nonexistent_returns_none(self): self.fm.create_user('alice', 'secret') result = self.fm.download_file('alice', 'nope.txt') self.assertIsNone(result) def test_delete_file_nonexistent_returns_false(self): self.fm.create_user('alice', 'secret') result = self.fm.delete_file('alice', 'nope.txt') self.assertFalse(result) def test_upload_file_creates_parent_dirs(self): self.fm.create_user('alice', 'secret') result = self.fm.upload_file('alice', 'deep/nested/file.txt', b'data') self.assertTrue(result) path = self.fm._safe_path('alice', 'deep/nested/file.txt') self.assertTrue(os.path.exists(path)) # ── list_files ────────────────────────────────────────────────────────── def test_list_files_shows_files_and_dirs(self): self.fm.create_user('alice', 'secret') self.fm.upload_file('alice', 'docs/readme.txt', b'hello') files = self.fm.list_files('alice', 'docs') self.assertEqual(len(files), 1) self.assertEqual(files[0]['type'], 'file') def test_list_files_empty_folder(self): self.fm.create_user('alice', 'secret') files = self.fm.list_files('alice', 'Videos') self.assertIsInstance(files, list) self.assertEqual(len(files), 0) # ── backup / restore ──────────────────────────────────────────────────── def test_backup_user_files_invalid_username_returns_false(self): result = self.fm.backup_user_files('../../etc', '/tmp/backup') self.assertFalse(result) def test_backup_user_files_empty_username_returns_false(self): result = self.fm.backup_user_files('', '/tmp/backup') self.assertFalse(result) def test_restore_user_files_invalid_username_returns_false(self): result = self.fm.restore_user_files('../../etc', '/tmp/backup') self.assertFalse(result) def test_restore_user_files_nonexistent_backup_returns_false(self): result = self.fm.restore_user_files('alice', '/tmp/nonexistent_backup') self.assertFalse(result) # ── _test_filesystem_access ───────────────────────────────────────────── def test_test_filesystem_access_succeeds(self): result = self.fm._test_filesystem_access() self.assertTrue(result['success']) self.assertTrue(result['read_write']) # ── _test_user_authentication ─────────────────────────────────────────── def test_test_user_authentication_no_file_returns_success(self): result = self.fm._test_user_authentication() self.assertTrue(result['success']) self.assertEqual(result['users_count'], 0) def test_test_user_authentication_counts_users(self): self.fm.create_user('alice', 'secret') self.fm.create_user('bob', 'secret2') result = self.fm._test_user_authentication() self.assertTrue(result['success']) self.assertEqual(result['users_count'], 2) # ── test_connectivity ─────────────────────────────────────────────────── @patch('requests.get') @patch('requests.options') def test_test_connectivity_returns_dict_with_expected_keys(self, mock_options, mock_get): mock_get.side_effect = Exception('connection refused') mock_options.side_effect = Exception('connection refused') result = self.fm.test_connectivity() self.assertIn('webdav_connectivity', result) self.assertIn('filesystem_access', result) self.assertIn('user_authentication', result) self.assertIn('success', result) # ── get_status ────────────────────────────────────────────────────────── @patch.dict(os.environ, {'DOCKER_CONTAINER': 'true'}) def test_get_status_docker_mode_returns_dict(self): result = self.fm.get_status() self.assertIn('running', result) self.assertIn('status', result) @patch.dict(os.environ, {'DOCKER_CONTAINER': 'false'}) @patch('subprocess.run') @patch('requests.get') @patch('requests.options') def test_get_status_non_docker_mode(self, mock_opt, mock_get, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='') mock_get.side_effect = Exception('refused') mock_opt.side_effect = Exception('refused') result = self.fm.get_status() self.assertIn('running', result) # ── _get_total_storage_used ───────────────────────────────────────────── def test_get_total_storage_used_empty_dir(self): result = self.fm._get_total_storage_used() self.assertEqual(result['total_files'], 0) self.assertEqual(result['total_size_bytes'], 0) def test_get_total_storage_used_with_files(self): self.fm.create_user('alice', 'secret') self.fm.upload_file('alice', 'Documents/test.txt', b'hello world') result = self.fm._get_total_storage_used() self.assertGreater(result['total_files'], 0) self.assertGreater(result['total_size_bytes'], 0) if __name__ == '__main__': unittest.main()