roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
65d6d07c8d4db73ea385bdab13ef7ff9d8e71825
pic/tests/test_file_endpoints.py
T
roof 44d7e96f29 feat: Phase 6 — require_active_service decorator + wizard install wiring 
Email/calendar/files routes now return 404 when the service is not
installed, using a require_active_service decorator that checks
ServiceRegistry. Status endpoints are exempt so health checks always work.

SetupManager.complete_setup() now accepts a service_store_manager and
installs any wizard-selected services in a background daemon thread after
setup completes. Failures are logged but do not fail the wizard.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 16:58:57 -04:00

339 lines
13 KiB
Python
Raw Blame History

#!/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/<username>/<path> (including path traversal)
GET /api/files/list/<username>
GET /api/files/download/<username>/<path>
DELETE /api/files/delete/<username>/<path>
POST /api/files/folders
POST /api/files/upload/<username>
"""
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
_INSTALLED = {'id': 'files', 'installed': True}
class TestFileUsersEndpoints(unittest.TestCase):
def setUp(self):
app.config['TESTING'] = True
self.client = app.test_client()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── 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()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── GET /api/files/list/<username> ─────────────────────────────────────
@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()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── DELETE /api/files/folders/<username>/<path> ────────────────────────
@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()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── GET /api/files/download/<username>/<path> ──────────────────────────
@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/<username>/<path> ─────────────────────────
@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()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── 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_path': '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_path': '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()
self._sr_patcher = patch('app.service_registry')
mock_sr = self._sr_patcher.start()
mock_sr.get.return_value = _INSTALLED
def tearDown(self):
self._sr_patcher.stop()
# ── POST /api/files/upload/<username> ──────────────────────────────────
@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()
Reference in New Issue View Git Blame Copy Permalink