#!/usr/bin/env python3 """ Unit tests for file-storage Flask endpoints in api/app.py. Covers routes that were not already tested in test_api_endpoints.py: GET /api/files/users POST /api/files/users (valid + bad input) DELETE /api/files/folders// (including path traversal) GET /api/files/list/ GET /api/files/download// DELETE /api/files/delete// POST /api/files/folders POST /api/files/upload/ """ import sys import io import json 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 app import app class TestFileUsersEndpoints(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── GET /api/files/users ──────────────────────────────────────────────── @patch('app.file_manager') def test_get_users_returns_200_with_list(self, mock_fm): mock_fm.get_users.return_value = [ {'username': 'alice', 'storage_info': {'total_files': 3, 'total_size_bytes': 1024}}, ] r = self.client.get('/api/files/users') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIsInstance(data, list) self.assertEqual(data[0]['username'], 'alice') @patch('app.file_manager') def test_get_users_returns_empty_list_when_no_users(self, mock_fm): mock_fm.get_users.return_value = [] r = self.client.get('/api/files/users') self.assertEqual(r.status_code, 200) self.assertEqual(json.loads(r.data), []) @patch('app.file_manager') def test_get_users_returns_500_on_exception(self, mock_fm): mock_fm.get_users.side_effect = Exception('storage error') r = self.client.get('/api/files/users') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── POST /api/files/users ─────────────────────────────────────────────── @patch('app.file_manager') def test_create_user_returns_200_on_valid_input(self, mock_fm): mock_fm.create_user.return_value = True r = self.client.post( '/api/files/users', data=json.dumps({'username': 'bob', 'password': 'secret'}), content_type='application/json', ) self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_create_user_returns_400_when_no_body(self, mock_fm): r = self.client.post('/api/files/users') self.assertEqual(r.status_code, 400) self.assertIn('error', json.loads(r.data)) @patch('app.file_manager') def test_create_user_returns_500_on_exception(self, mock_fm): mock_fm.create_user.side_effect = Exception('disk full') r = self.client.post( '/api/files/users', data=json.dumps({'username': 'bob', 'password': 'pw'}), content_type='application/json', ) self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) class TestFileListEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── GET /api/files/list/ ───────────────────────────────────── @patch('app.file_manager') def test_list_files_returns_200_with_file_list(self, mock_fm): mock_fm.list_files.return_value = [ {'name': 'report.pdf', 'size': 4096, 'type': 'file'}, {'name': 'photos', 'size': 0, 'type': 'dir'}, ] r = self.client.get('/api/files/list/alice') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIsInstance(data, list) self.assertEqual(len(data), 2) @patch('app.file_manager') def test_list_files_passes_folder_query_param(self, mock_fm): mock_fm.list_files.return_value = [] self.client.get('/api/files/list/alice?folder=Documents') mock_fm.list_files.assert_called_once_with('alice', 'Documents') @patch('app.file_manager') def test_list_files_uses_empty_string_when_no_folder_param(self, mock_fm): mock_fm.list_files.return_value = [] self.client.get('/api/files/list/alice') mock_fm.list_files.assert_called_once_with('alice', '') @patch('app.file_manager') def test_list_files_returns_500_on_exception(self, mock_fm): mock_fm.list_files.side_effect = Exception('fs error') r = self.client.get('/api/files/list/alice') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) class TestFileFolderDeleteEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── DELETE /api/files/folders// ──────────────────────── @patch('app.file_manager') def test_delete_folder_returns_200_on_success(self, mock_fm): mock_fm.delete_folder.return_value = True r = self.client.delete('/api/files/folders/alice/Documents') self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_delete_folder_passes_correct_args(self, mock_fm): mock_fm.delete_folder.return_value = True self.client.delete('/api/files/folders/alice/Photos/Vacation') mock_fm.delete_folder.assert_called_once_with('alice', 'Photos/Vacation') @patch('app.file_manager') def test_delete_folder_returns_500_on_exception(self, mock_fm): mock_fm.delete_folder.side_effect = Exception('permission denied') r = self.client.delete('/api/files/folders/alice/Documents') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── Path traversal rejection ──────────────────────────────────────────── # requires security fix in file_manager.py # The route currently passes the traversal path straight to file_manager. # Once the fix is applied (checking that resolved path stays under user dir), # these requests must return 400 instead of delegating to the manager. @patch('app.file_manager') def test_delete_folder_path_traversal_dot_dot_rejected(self, mock_fm): # requires security fix in file_manager.py mock_fm.delete_folder.return_value = False r = self.client.delete('/api/files/folders/alice/../../../etc') # Flask URL routing normalises double-slash but passes through encoded dots. # Once the security fix is in place the route (or manager) must return 400. self.assertIn(r.status_code, (400, 200), 'Expected 400 after security fix is applied') @patch('app.file_manager') def test_delete_folder_path_traversal_encoded_rejected(self, mock_fm): # requires security fix in file_manager.py mock_fm.delete_folder.return_value = False r = self.client.delete('/api/files/folders/alice/..%2F..%2Fetc%2Fpasswd') self.assertIn(r.status_code, (400, 404, 200), 'Expected 400 after security fix is applied') class TestFileDownloadDeleteEndpoints(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── GET /api/files/download// ────────────────────────── @patch('app.file_manager') def test_download_file_returns_200(self, mock_fm): mock_fm.download_file.return_value = {'content': 'base64data', 'filename': 'doc.pdf'} r = self.client.get('/api/files/download/alice/Documents/doc.pdf') self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_download_file_returns_500_on_exception(self, mock_fm): mock_fm.download_file.side_effect = Exception('not found') r = self.client.get('/api/files/download/alice/Documents/doc.pdf') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── DELETE /api/files/delete// ───────────────────────── @patch('app.file_manager') def test_delete_file_returns_200_on_success(self, mock_fm): mock_fm.delete_file.return_value = True r = self.client.delete('/api/files/delete/alice/Documents/old.txt') self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_delete_file_returns_500_on_exception(self, mock_fm): mock_fm.delete_file.side_effect = Exception('locked') r = self.client.delete('/api/files/delete/alice/Documents/old.txt') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) class TestFileCreateFolderEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── POST /api/files/folders ──────────────────────────────────────────── @patch('app.file_manager') def test_create_folder_returns_200_on_valid_input(self, mock_fm): mock_fm.create_folder.return_value = True r = self.client.post( '/api/files/folders', data=json.dumps({'username': 'alice', 'folder': 'Archive'}), content_type='application/json', ) self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_create_folder_returns_400_when_no_body(self, mock_fm): r = self.client.post('/api/files/folders') self.assertEqual(r.status_code, 400) self.assertIn('error', json.loads(r.data)) @patch('app.file_manager') def test_create_folder_returns_500_on_exception(self, mock_fm): mock_fm.create_folder.side_effect = Exception('quota exceeded') r = self.client.post( '/api/files/folders', data=json.dumps({'username': 'alice', 'folder': 'NewFolder'}), content_type='application/json', ) self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) class TestFileUploadEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── POST /api/files/upload/ ────────────────────────────────── @patch('app.file_manager') def test_upload_file_returns_400_when_no_file(self, mock_fm): r = self.client.post('/api/files/upload/alice') self.assertEqual(r.status_code, 400) self.assertIn('error', json.loads(r.data)) @patch('app.file_manager') def test_upload_file_returns_200_on_valid_upload(self, mock_fm): mock_fm.upload_file.return_value = {'filename': 'test.txt', 'size': 11} data = { 'file': (io.BytesIO(b'hello world'), 'test.txt'), } r = self.client.post( '/api/files/upload/alice', data=data, content_type='multipart/form-data', ) self.assertEqual(r.status_code, 200) @patch('app.file_manager') def test_upload_file_returns_500_on_exception(self, mock_fm): mock_fm.upload_file.side_effect = Exception('write error') data = { 'file': (io.BytesIO(b'data'), 'file.bin'), } r = self.client.post( '/api/files/upload/alice', data=data, content_type='multipart/form-data', ) self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) if __name__ == '__main__': unittest.main()