44d7e96f29
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>
215 lines
8.6 KiB
Python
215 lines
8.6 KiB
Python
"""Tests for the require_active_service route decorator."""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
from flask import Flask
|
|
from routes import require_active_service
|
|
|
|
|
|
class TestRequireActiveServiceDecorator(unittest.TestCase):
|
|
|
|
def test_installed_service_passes_through(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {'id': 'email'}
|
|
|
|
@app.route('/test/email')
|
|
@require_active_service('email')
|
|
def view():
|
|
return 'ok', 200
|
|
|
|
with app.test_client() as client:
|
|
with patch.dict('sys.modules', {'app': MagicMock(service_registry=mock_registry)}):
|
|
resp = client.get('/test/email')
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
def test_not_installed_returns_404(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
|
|
@app.route('/test/calendar')
|
|
@require_active_service('calendar')
|
|
def view():
|
|
return 'ok', 200
|
|
|
|
with app.test_client() as client:
|
|
with patch.dict('sys.modules', {'app': MagicMock(service_registry=mock_registry)}):
|
|
resp = client.get('/test/calendar')
|
|
self.assertEqual(resp.status_code, 404)
|
|
data = resp.get_json()
|
|
self.assertIn('error', data)
|
|
self.assertIn('calendar', data['error'])
|
|
|
|
def test_not_installed_error_message_contains_service_id(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
|
|
@app.route('/test/files')
|
|
@require_active_service('files')
|
|
def view():
|
|
return 'ok', 200
|
|
|
|
with app.test_client() as client:
|
|
with patch.dict('sys.modules', {'app': MagicMock(service_registry=mock_registry)}):
|
|
resp = client.get('/test/files')
|
|
data = resp.get_json()
|
|
self.assertIn('files', data['error'])
|
|
|
|
def test_decorator_preserves_function_name(self):
|
|
@require_active_service('calendar')
|
|
def my_view():
|
|
return 'ok'
|
|
self.assertEqual(my_view.__name__, 'my_view')
|
|
|
|
def test_decorator_preserves_function_docstring(self):
|
|
@require_active_service('email')
|
|
def documented_view():
|
|
"""Returns email data."""
|
|
return 'ok'
|
|
self.assertEqual(documented_view.__doc__, 'Returns email data.')
|
|
|
|
def test_passes_positional_args_to_wrapped_function(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {'id': 'email'}
|
|
|
|
@app.route('/test/mailbox/<username>')
|
|
@require_active_service('email')
|
|
def mailbox_view(username):
|
|
return username, 200
|
|
|
|
with app.test_client() as client:
|
|
with patch.dict('sys.modules', {'app': MagicMock(service_registry=mock_registry)}):
|
|
resp = client.get('/test/mailbox/alice')
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b'alice', resp.data)
|
|
|
|
def test_service_registry_get_called_with_correct_service_id(self):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {'id': 'files'}
|
|
|
|
@app.route('/test/svc')
|
|
@require_active_service('files')
|
|
def view():
|
|
return 'ok', 200
|
|
|
|
with app.test_client() as client:
|
|
with patch.dict('sys.modules', {'app': MagicMock(service_registry=mock_registry)}):
|
|
client.get('/test/svc')
|
|
mock_registry.get.assert_called_with('files')
|
|
|
|
def test_status_endpoint_bypasses_decorator_on_email_routes(self):
|
|
"""The /status route in email.py must NOT be decorated; verify by importing the route."""
|
|
from routes.email import get_email_status
|
|
# The status handler should not be wrapped — it won't have the
|
|
# _require_active_service marker that the wrapper would add.
|
|
# We verify by checking it has no 'service_id' closure variable
|
|
# from the decorator (i.e., it's the plain function, not a wrapper).
|
|
import inspect
|
|
# get_email_status should have no closure cells referencing a service_id
|
|
closure = get_email_status.__closure__
|
|
if closure:
|
|
closure_vals = [c.cell_contents for c in closure if isinstance(c.cell_contents, str)]
|
|
self.assertNotIn('email', closure_vals,
|
|
"get_email_status should not be wrapped by require_active_service")
|
|
|
|
def test_status_endpoint_bypasses_decorator_on_calendar_routes(self):
|
|
from routes.calendar import get_calendar_status
|
|
closure = get_calendar_status.__closure__
|
|
if closure:
|
|
closure_vals = [c.cell_contents for c in closure if isinstance(c.cell_contents, str)]
|
|
self.assertNotIn('calendar', closure_vals)
|
|
|
|
def test_status_endpoint_bypasses_decorator_on_files_routes(self):
|
|
from routes.files import get_file_status
|
|
closure = get_file_status.__closure__
|
|
if closure:
|
|
closure_vals = [c.cell_contents for c in closure if isinstance(c.cell_contents, str)]
|
|
self.assertNotIn('files', closure_vals)
|
|
|
|
|
|
class TestRequireActiveServiceOnEmailRoutes(unittest.TestCase):
|
|
"""Integration-style: exercise the decorator via the real Flask app."""
|
|
|
|
def setUp(self):
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
from app import app
|
|
app.config['TESTING'] = True
|
|
self.client = app.test_client()
|
|
self.app = app
|
|
|
|
def test_email_users_returns_404_when_service_not_installed(self):
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
with patch('app.service_registry', mock_registry):
|
|
resp = self.client.get('/api/email/users')
|
|
self.assertEqual(resp.status_code, 404)
|
|
data = resp.get_json()
|
|
self.assertIn('error', data)
|
|
self.assertIn('email', data['error'])
|
|
|
|
def test_email_status_reachable_when_service_not_installed(self):
|
|
"""Status endpoint is never blocked, even with service absent."""
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
mock_email_mgr = MagicMock()
|
|
mock_email_mgr.get_status.return_value = {'installed': False}
|
|
with patch('app.service_registry', mock_registry), \
|
|
patch('app.email_manager', mock_email_mgr):
|
|
resp = self.client.get('/api/email/status')
|
|
self.assertNotEqual(resp.status_code, 404)
|
|
|
|
def test_calendar_users_returns_404_when_service_not_installed(self):
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
with patch('app.service_registry', mock_registry):
|
|
resp = self.client.get('/api/calendar/users')
|
|
self.assertEqual(resp.status_code, 404)
|
|
data = resp.get_json()
|
|
self.assertIn('calendar', data['error'])
|
|
|
|
def test_calendar_status_reachable_when_service_not_installed(self):
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
mock_cal_mgr = MagicMock()
|
|
mock_cal_mgr.get_status.return_value = {'installed': False}
|
|
with patch('app.service_registry', mock_registry), \
|
|
patch('app.calendar_manager', mock_cal_mgr):
|
|
resp = self.client.get('/api/calendar/status')
|
|
self.assertNotEqual(resp.status_code, 404)
|
|
|
|
def test_files_users_returns_404_when_service_not_installed(self):
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
with patch('app.service_registry', mock_registry):
|
|
resp = self.client.get('/api/files/users')
|
|
self.assertEqual(resp.status_code, 404)
|
|
data = resp.get_json()
|
|
self.assertIn('files', data['error'])
|
|
|
|
def test_files_status_reachable_when_service_not_installed(self):
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
mock_file_mgr = MagicMock()
|
|
mock_file_mgr.get_status.return_value = {'installed': False}
|
|
with patch('app.service_registry', mock_registry), \
|
|
patch('app.file_manager', mock_file_mgr):
|
|
resp = self.client.get('/api/files/status')
|
|
self.assertNotEqual(resp.status_code, 404)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|