111 lines
7.1 KiB
Python
111 lines
7.1 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
from backend.app.config import settings
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from backend.app import db
|
|
from backend.app.feature_access import FEATURES, permissions, update_permissions
|
|
from backend.app.routers import admin, auth, events, insights, portal, recaps, requests
|
|
from backend.app.security import create_access_token
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
|
|
|
|
class FeatureAccessTests(TempDatabaseMixin, unittest.TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
secret = patch.object(settings, "jwt_secret", "feature-access-tests-only-secret-123456789")
|
|
secret.start()
|
|
self.addCleanup(secret.stop)
|
|
db.create_user('feature-viewer', 'Example-password123!', role='user')
|
|
db.create_user('feature-admin', 'Example-password123!', role='admin')
|
|
self.user = db.get_user_by_username('feature-viewer')
|
|
app = FastAPI()
|
|
for module in (admin, auth, events, insights, portal, recaps, requests):
|
|
app.include_router(module.router)
|
|
self.client = TestClient(app)
|
|
self.client.headers['Authorization'] = 'Bearer ' + create_access_token(self.user['username'], 'user')
|
|
|
|
def test_defaults_persist_and_invites_share_existing_setting(self):
|
|
self.assertEqual(permissions(self.user), dict(stats=True, requests=True, new_requests=True, issues=True, invites=False))
|
|
update_permissions({'stats': False, 'invites': True}, self.user['username'])
|
|
db.init_db()
|
|
fresh = db.get_user_by_username(self.user['username'])
|
|
self.assertTrue(fresh['invite_management_enabled'])
|
|
self.assertFalse(permissions(fresh)['stats'])
|
|
db.set_user_invite_management_enabled(self.user['username'], False)
|
|
self.assertFalse(permissions(db.get_user_by_username(self.user['username']))['invites'])
|
|
|
|
def test_all_feature_apis_reject_disabled_access_with_existing_token(self):
|
|
update_permissions(dict.fromkeys(FEATURES, False), self.user['username'])
|
|
endpoints = [
|
|
('GET', '/insights', None), ('GET', '/insights/reports/monthly', None),
|
|
('GET', '/insights/reports/monthly.csv', None), ('GET', '/insights/artwork/item?token=x', None),
|
|
('GET', '/profile/email-recaps', None), ('POST', '/profile/email-recaps/send', {}),
|
|
('GET', '/requests/recent', None), ('GET', '/requests/search?query=Movie', None),
|
|
('GET', '/requests/request-options?mediaType=movie&tmdbId=1', None),
|
|
('POST', '/requests/create', {'mediaType': 'movie', 'tmdbId': 1}),
|
|
('GET', '/requests/1/snapshot', None), ('POST', '/requests/1/actions/search', {}),
|
|
('GET', '/requests/1/issue-options', None), ('POST', '/requests/1/actions/replace', {}),
|
|
('GET', '/portal/items?kind=issue', None), ('GET', '/portal/requests', None),
|
|
('POST', '/portal/items', {'kind': 'issue'}), ('POST', '/portal/items', {'kind': 'request'}),
|
|
('GET', '/portal/issues/media-status', None), ('POST', '/portal/requests/1/issues', {}),
|
|
('GET', '/auth/profile/invites', None), ('POST', '/auth/profile/invites', {}),
|
|
('PUT', '/auth/profile/invites/1', {}), ('DELETE', '/auth/profile/invites/1', None),
|
|
('GET', '/events/stream', None), ('GET', '/events/requests/1/stream', None),
|
|
]
|
|
for method, path, payload in endpoints:
|
|
with self.subTest(path=path, method=method):
|
|
self.assertEqual(self.client.request(method, path, json=payload).status_code, 403)
|
|
self.assertEqual(self.client.get('/auth/me').json()['features'], dict.fromkeys(FEATURES, False))
|
|
self.assertEqual(self.client.get('/auth/profile').status_code, 200)
|
|
|
|
def test_bulk_is_admin_only_strict_and_leaves_other_features_untouched(self):
|
|
self.assertEqual(self.client.put('/admin/users/features/bulk', json={'issues': False}).status_code, 403)
|
|
self.client.headers['Authorization'] = 'Bearer ' + create_access_token('feature-admin', 'admin')
|
|
for invalid in ({'issues': 'false'}, {'unknown': True}, {}):
|
|
self.assertEqual(self.client.put('/admin/users/features/bulk', json=invalid).status_code, 400)
|
|
response = self.client.put('/admin/users/features/bulk', json={'issues': False})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()['updated'], 1)
|
|
self.assertFalse(permissions(self.user)['issues'])
|
|
self.assertTrue(permissions(self.user)['requests'])
|
|
self.assertTrue(all(permissions(db.get_user_by_username('feature-admin')).values()))
|
|
self.assertEqual(self.client.put('/admin/users/feature-admin/features', json={'stats': False}).status_code, 400)
|
|
self.assertEqual(self.client.put('/admin/users/missing/features', json={'stats': False}).status_code, 404)
|
|
|
|
def test_issue_and_request_item_routes_cannot_bypass_disabled_feature(self):
|
|
issue = db.create_portal_item(kind='issue', title='Problem', description='Problem', created_by_username=self.user['username'], created_by_id=self.user['id'])
|
|
update_permissions({'issues': False}, self.user['username'])
|
|
for path in (f'/portal/items/{issue["id"]}', f'/portal/items/{issue["id"]}/comments', '/portal/items', '/portal/overview'):
|
|
self.assertEqual(self.client.get(path).status_code, 403)
|
|
self.assertEqual(self.client.get('/portal/requests').status_code, 200)
|
|
self.assertEqual(self.client.get('/portal/items?kind=request').status_code, 200)
|
|
update_permissions({'issues': True, 'requests': False, 'new_requests': False}, self.user['username'])
|
|
self.assertEqual(self.client.get(f'/portal/items/{issue["id"]}').status_code, 200)
|
|
self.assertEqual(self.client.get('/portal/items?kind=issue').status_code, 200)
|
|
overview = self.client.get('/portal/overview?kind=issue')
|
|
self.assertEqual(overview.status_code, 200)
|
|
self.assertEqual(overview.json()['overview']['by_kind'], {'issue': 1})
|
|
self.assertEqual(self.client.post('/requests/create', json={'mediaType': 'movie', 'tmdbId': 1}).status_code, 403)
|
|
|
|
def test_deleted_account_does_not_leave_permissions_for_reused_id(self):
|
|
update_permissions({'stats': False}, self.user['username'])
|
|
db.delete_user_by_username(self.user['username'])
|
|
with db._connect() as conn:
|
|
self.assertEqual(conn.execute('SELECT COUNT(*) FROM user_feature_permissions').fetchone()[0], 0)
|
|
|
|
|
|
def test_open_request_stream_closes_after_permission_revocation(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock
|
|
from types import SimpleNamespace
|
|
async def scenario():
|
|
request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
|
|
response = await events.events_stream(request, user={**self.user, "features": permissions(self.user)})
|
|
iterator = response.body_iterator
|
|
self.assertIn('retry', await anext(iterator))
|
|
update_permissions({'requests': False}, self.user['username'])
|
|
with self.assertRaises(StopAsyncIteration):
|
|
await anext(iterator)
|
|
asyncio.run(scenario())
|