217 lines
9.6 KiB
Python
217 lines
9.6 KiB
Python
# DRF Rack
|
|
# Библиотека классов расширений для Django REST Framework
|
|
#
|
|
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
|
|
# Все права защищены. Лицензия: MIT
|
|
|
|
"""
|
|
Описание класса SessionManagerTest
|
|
"""
|
|
|
|
import datetime
|
|
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
from django.test import override_settings
|
|
from django.utils import timezone
|
|
|
|
from net.xeaf.rack.core import CoreTestCase
|
|
from net.xeaf.rack.enums import AuthRejectReason
|
|
from net.xeaf.rack.management.userstory import AccountTestData
|
|
from net.xeaf.rack.management.userstory import SessionTestData
|
|
from net.xeaf.rack.managers import SessionManager
|
|
from net.xeaf.rack.models import AccountModel
|
|
from net.xeaf.rack.models import SessionModel
|
|
from net.xeaf.rack.utils.exceptions import UnauthorizedException
|
|
|
|
|
|
class SessionManagerTests(CoreTestCase):
|
|
"""
|
|
Тесты для класса SessionManager
|
|
"""
|
|
|
|
def test_should_successfully_authorize_ivanov_with_real_password(self) -> None:
|
|
"""
|
|
Проверка успешного создания сессии для эталонного аккаунта
|
|
"""
|
|
|
|
session = SessionManager.authenticate(
|
|
username="i.ivanov@xeaf.net",
|
|
password=AccountTestData.DEFAULT_PASSWORD,
|
|
device_id="mobile_app"
|
|
)
|
|
|
|
# Проверяем, что вернулась живая сессия со сгенерированным токеном
|
|
self.assertIsNotNone(session)
|
|
self.assertEqual(len(session.token), 40)
|
|
self.assertEqual(session.account_id, AccountTestData.I_IVANOV_ID)
|
|
self.assertEqual(session.actual_account_id, AccountTestData.I_IVANOV_ID)
|
|
self.assertEqual(session.device_id, "mobile_app")
|
|
|
|
# Честно проверяем запись физически внутри БД
|
|
db_session = SessionModel.objects.get(token=session.token)
|
|
self.assertEqual(db_session.account_id, AccountTestData.I_IVANOV_ID)
|
|
|
|
def test_should_raise_error_when_password_is_incorrect(self) -> None:
|
|
"""
|
|
Менеджер должен выкинуть исключение ядра при неверном пароле
|
|
"""
|
|
|
|
with self.assertRaises(UnauthorizedException) as context:
|
|
SessionManager.authenticate(
|
|
username="i.ivanov@xeaf.net",
|
|
password="wrong_password_here"
|
|
)
|
|
|
|
self.assertIsInstance(context.exception.meta, dict)
|
|
self.assertEqual(context.exception.meta['reason'], AuthRejectReason.INVALID_USERNAME_OR_PASSWORD)
|
|
|
|
def test_should_raise_error_when_username_does_not_exist(self) -> None:
|
|
"""
|
|
Менеджер должен выкинуть исключение ядра, если email не найден
|
|
"""
|
|
|
|
with self.assertRaises(UnauthorizedException) as context:
|
|
SessionManager.authenticate(
|
|
username="unknown.user@xeaf.net",
|
|
password=AccountTestData.DEFAULT_PASSWORD
|
|
)
|
|
|
|
self.assertIsInstance(context.exception.meta, dict)
|
|
self.assertEqual(context.exception.meta['reason'], AuthRejectReason.INVALID_USERNAME_OR_PASSWORD)
|
|
|
|
def test_should_raise_error_if_account_is_inactive(self) -> None:
|
|
"""
|
|
Проверка деструктивного изменения БД: отключение аккаунта блокирует вход
|
|
"""
|
|
|
|
petrov = AccountModel.objects.get(id=AccountTestData.P_PETROV_ID)
|
|
petrov.is_active = False
|
|
petrov.save()
|
|
|
|
with self.assertRaises(UnauthorizedException) as context:
|
|
SessionManager.authenticate(
|
|
username="p.petrov@xeaf.net",
|
|
password=AccountTestData.DEFAULT_PASSWORD
|
|
)
|
|
|
|
self.assertIsInstance(context.exception.meta, dict)
|
|
self.assertEqual(context.exception.meta['reason'], AuthRejectReason.ACCOUNT_IS_NOT_ACTIVE)
|
|
|
|
def test_should_find_real_ivanov_session_by_token(self) -> None:
|
|
"""
|
|
Поиск существующей и активной сессии по токену из фикстуры
|
|
"""
|
|
|
|
session = SessionManager.find_by_token(SessionTestData.I_IVANOV_SESSION_TOKEN)
|
|
|
|
self.assertIsNotNone(session)
|
|
self.assertIsInstance(session, SessionModel)
|
|
self.assertEqual(session.token, SessionTestData.I_IVANOV_SESSION_TOKEN)
|
|
self.assertEqual(session.account_id, AccountTestData.I_IVANOV_ID)
|
|
|
|
def test_should_return_none_for_non_existent_token(self) -> None:
|
|
"""
|
|
Запрос несуществующего токена должен возвращать None
|
|
"""
|
|
|
|
result = SessionManager.find_by_token("non_existent_token_garbage_data")
|
|
self.assertIsNone(result)
|
|
|
|
def test_should_return_none_if_session_is_expired_in_db(self) -> None:
|
|
"""
|
|
Проверка деструктивного изменения БД: просроченная сессия не считывается
|
|
"""
|
|
|
|
session = SessionModel.objects.get(token=SessionTestData.P_PETROV_SESSION_TOKEN)
|
|
|
|
# Искусственно сдвигаем дату экспирации в далекое прошлое прямо в базе данных
|
|
session.expires_at = timezone.now() - datetime.timedelta(days=10)
|
|
session.save()
|
|
|
|
result = SessionManager.find_by_token(SessionTestData.P_PETROV_SESSION_TOKEN)
|
|
self.assertIsNone(result)
|
|
|
|
def test_should_fail_finding_if_session_was_physically_deleted_from_db(self) -> None:
|
|
"""
|
|
Полное физическое удаление строки из БД корректно обрабатывается
|
|
"""
|
|
|
|
SessionModel.objects.filter(token=SessionTestData.I_IVANOV_SESSION_TOKEN).delete()
|
|
|
|
result = SessionManager.find_by_token(SessionTestData.I_IVANOV_SESSION_TOKEN)
|
|
self.assertIsNone(result)
|
|
|
|
@override_settings(SESSION_TTL=1800)
|
|
def test_should_open_session_and_apply_override_settings_ttl(self) -> None:
|
|
"""
|
|
Открытие сессии должно использовать живой ConfigurationManager без заглушек
|
|
"""
|
|
|
|
ivanov = AccountModel.objects.get(id=AccountTestData.I_IVANOV_ID)
|
|
petrov = AccountModel.objects.get(id=AccountTestData.P_PETROV_ID)
|
|
|
|
# Создаем сессию имперсонации (Иванов под видом Петра)
|
|
session = SessionManager.open_session(account=ivanov, actual_account=petrov, device_id="tablet_v1")
|
|
|
|
# Проверяем, что запись успешно материализовалась в ОЗУ
|
|
db_session = SessionModel.objects.get(token=session.token)
|
|
self.assertEqual(db_session.account, ivanov)
|
|
self.assertEqual(db_session.actual_account, petrov)
|
|
self.assertEqual(db_session.device_id, "tablet_v1")
|
|
self.assertGreater(db_session.expires_at, timezone.now())
|
|
|
|
def test_should_physically_expire_session_on_close_session(self) -> None:
|
|
"""
|
|
Закрытие сессии переводит модель в статус expired прямо в БД
|
|
"""
|
|
|
|
session = SessionModel.objects.get(token=SessionTestData.I_IVANOV_SESSION_TOKEN)
|
|
|
|
SessionManager.close_session(session)
|
|
|
|
# Перечитываем чистую запись из БД
|
|
db_session = SessionModel.objects.get(token=SessionTestData.I_IVANOV_SESSION_TOKEN)
|
|
self.assertTrue(db_session.is_expired())
|
|
|
|
@override_settings(SESSION_TTL=9999)
|
|
def test_should_extend_expiration_time_on_renew_session(self) -> None:
|
|
"""
|
|
Обновление сессии сдвигает expires_at вперед на величину из настроек
|
|
"""
|
|
|
|
session = SessionModel.objects.get(token=SessionTestData.P_PETROV_SESSION_TOKEN)
|
|
initial_expires_at = session.expires_at
|
|
|
|
SessionManager.renew_session(session)
|
|
|
|
db_session = SessionModel.objects.get(token=SessionTestData.P_PETROV_SESSION_TOKEN)
|
|
self.assertGreater(db_session.expires_at, initial_expires_at)
|
|
|
|
def test_protected_create_session_model_writes_correct_record(self) -> None:
|
|
"""
|
|
Прямой вызов защищенного метода фабрики для хирургического теста записи
|
|
"""
|
|
|
|
ivanov = AccountModel.objects.get(id=AccountTestData.I_IVANOV_ID)
|
|
|
|
# Запускаем защищенный метод фабрики в обход цепочек обновления TTL
|
|
session = SessionManager._create_session_model(account=ivanov, actual_account=ivanov)
|
|
|
|
# Доказываем линтеру, что мы получили объект, и проверяем его наличие в БД
|
|
self.assertIsInstance(session, SessionModel)
|
|
self.assertTrue(SessionModel.objects.filter(token=session.token).exists())
|
|
|
|
@override_settings(AUTH_SESSION_MODEL="invalid_app.MissingModel")
|
|
def test_protected_get_session_model_class_raises_on_missing_model(self) -> None:
|
|
"""
|
|
Защищенный метод разрешения класса должен падать, если модель не найдена
|
|
"""
|
|
|
|
# Сбрасываем внутренний кэш класса в синглтоне менеджера перед тестом
|
|
SessionManager._SessionManager__model_class = None
|
|
|
|
with self.assertRaises(ImproperlyConfigured) as context:
|
|
SessionManager._get_session_model_class()
|
|
|
|
self.assertIn("must be set in settings.py", str(context.exception))
|