55 lines
3.2 KiB
Python
55 lines
3.2 KiB
Python
import asyncio
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
from fastapi import HTTPException, Response
|
|
from backend.app import db
|
|
from backend.app.routers import auth
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
|
|
|
|
class InviteEmailSignupTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
async def signup(self, code, username, **extra):
|
|
with patch.object(auth, 'get_runtime_settings', return_value=SimpleNamespace(jellyfin_base_url=None, jellyfin_api_key=None)), patch.object(auth, 'send_templated_email', new_callable=AsyncMock), patch.object(auth, 'create_access_token', return_value='test-token'):
|
|
return await auth.signup({'invite_code': code, 'username': username, 'password': 'Strong-Test-Password123!', **extra}, Response())
|
|
|
|
async def test_email_invite_binds_account_and_cannot_be_reused(self):
|
|
invite = db.create_signup_invite(code='EMAILTEST', recipient_email='recipient@example.com', max_uses=20)
|
|
self.assertEqual(invite['max_uses'], 1)
|
|
public = auth._public_invite_payload(invite)
|
|
self.assertTrue(public['email_bound'])
|
|
self.assertNotIn('recipient@example.com', str(public))
|
|
await self.signup('EMAILTEST', 'first-user')
|
|
self.assertEqual(db.get_user_by_username('first-user')['email'], 'recipient@example.com')
|
|
with self.assertRaises(HTTPException):
|
|
await self.signup('EMAILTEST', 'second-user')
|
|
|
|
async def test_email_invite_rejects_recipient_override(self):
|
|
db.create_signup_invite(code='BOUNDTEST', recipient_email='recipient@example.com')
|
|
with self.assertRaises(HTTPException):
|
|
await self.signup('BOUNDTEST', 'override-user', email='different@example.com')
|
|
self.assertEqual(db.get_signup_invite_by_code('BOUNDTEST')['use_count'], 0)
|
|
|
|
async def test_manual_invite_requires_and_saves_email(self):
|
|
db.create_signup_invite(code='MANUALTEST', max_uses=3)
|
|
for email in ['', 'invalid']:
|
|
with self.assertRaises(HTTPException):
|
|
await self.signup('MANUALTEST', 'manual-user', email=email)
|
|
await self.signup('MANUALTEST', 'manual-user', email='manual@example.com')
|
|
self.assertEqual(db.get_user_by_username('manual-user')['email'], 'manual@example.com')
|
|
self.assertEqual(db.get_signup_invite_by_code('MANUALTEST')['remaining_uses'], 2)
|
|
|
|
async def test_failed_creation_releases_reservation(self):
|
|
invite = db.create_signup_invite(code='FAILTEST', recipient_email='recipient@example.com')
|
|
with patch.object(auth, 'create_user', side_effect=RuntimeError('test failure')):
|
|
with self.assertRaises(HTTPException):
|
|
await self.signup('FAILTEST', 'failed-user')
|
|
self.assertEqual(db.get_signup_invite_by_id(invite['id'])['use_count'], 0)
|
|
|
|
async def test_single_use_reservation_is_atomic(self):
|
|
invite = db.create_signup_invite(code='RACETEST', recipient_email='recipient@example.com')
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(db.reserve_signup_invite_use, [invite['id']] * 4))
|
|
self.assertEqual(sum(results), 1)
|