from __future__ import annotations import logging from collections.abc import Callable import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from config.database import Base from scripts import initial_data from scripts.initial_data import AdminSeedResult, RoleSeedResult, SeedConfig from services.repositories import DEFAULT_ROLE_DEFINITIONS from services.unit_of_work import UnitOfWork @pytest.fixture() def in_memory_session_factory() -> Callable[[], Session]: engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) factory = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) def _session_factory() -> Session: return factory() return _session_factory @pytest.fixture() def uow(in_memory_session_factory: Callable[[], Session]) -> UnitOfWork: return UnitOfWork(session_factory=in_memory_session_factory) def test_ensure_default_roles_idempotent(uow: UnitOfWork) -> None: with uow as working: assert working.roles is not None result_first = initial_data.ensure_default_roles(working.roles, DEFAULT_ROLE_DEFINITIONS) assert result_first == RoleSeedResult(created=4, updated=0, total=4) with uow as working: assert working.roles is not None result_second = initial_data.ensure_default_roles(working.roles, DEFAULT_ROLE_DEFINITIONS) assert result_second == RoleSeedResult(created=0, updated=0, total=4) def test_ensure_admin_user_creates_and_assigns_roles(uow: UnitOfWork) -> None: config = SeedConfig( admin_email="admin@example.com", admin_username="admin", admin_password="secret", admin_roles=("admin", "viewer"), force_reset=False, ) with uow as working: assert working.roles is not None assert working.users is not None initial_data.ensure_default_roles(working.roles, DEFAULT_ROLE_DEFINITIONS) result = initial_data.ensure_admin_user(working.users, working.roles, config) assert result == AdminSeedResult( created_user=True, updated_user=False, password_rotated=False, roles_granted=2, ) with uow as working: assert working.roles is not None assert working.users is not None result_again = initial_data.ensure_admin_user(working.users, working.roles, config) assert result_again == AdminSeedResult( created_user=False, updated_user=False, password_rotated=False, roles_granted=0, ) with uow as working: assert working.users is not None user = working.users.get_by_email("admin@example.com", with_roles=True) assert user is not None assert user.is_active is True assert user.is_superuser is True role_names = {role.name for role in user.roles} assert role_names == {"admin", "viewer"} def test_ensure_admin_user_force_reset_rotates_password(uow: UnitOfWork) -> None: base_config = SeedConfig( admin_email="admin@example.com", admin_username="admin", admin_password="first", admin_roles=("admin",), force_reset=False, ) with uow as working: assert working.roles is not None assert working.users is not None initial_data.ensure_default_roles(working.roles, DEFAULT_ROLE_DEFINITIONS) initial_data.ensure_admin_user(working.users, working.roles, base_config) rotate_config = SeedConfig( admin_email="admin@example.com", admin_username="admin", admin_password="second", admin_roles=("admin",), force_reset=True, ) with uow as working: assert working.users is not None user_before = working.users.get_by_email("admin@example.com") assert user_before is not None old_hash = user_before.password_hash with uow as working: assert working.roles is not None assert working.users is not None result = initial_data.ensure_admin_user(working.users, working.roles, rotate_config) assert result.password_rotated is True with uow as working: assert working.users is not None user_after = working.users.get_by_email("admin@example.com") assert user_after is not None assert user_after.password_hash != old_hash def test_seed_initial_data_logs_results( caplog, in_memory_session_factory: Callable[[], Session], ) -> None: caplog.set_level(logging.INFO) config = SeedConfig( admin_email="seed@example.com", admin_username="seed", admin_password="seed-pass", admin_roles=("admin",), force_reset=False, ) initial_data.seed_initial_data( config, unit_of_work_factory=lambda: UnitOfWork(session_factory=in_memory_session_factory), ) assert "Starting initial data seeding" in caplog.text assert "Initial data seeding completed successfully" in caplog.text with UnitOfWork(session_factory=in_memory_session_factory) as check_uow: assert check_uow.users is not None assert check_uow.roles is not None user = check_uow.users.get_by_email("seed@example.com") assert user is not None assert check_uow.roles.get_by_name("admin") is not None