117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from config.database import Base
|
|
from config.settings import AdminBootstrapSettings
|
|
from services.bootstrap import AdminBootstrapResult, RoleBootstrapResult, bootstrap_admin
|
|
from services.unit_of_work import UnitOfWork
|
|
|
|
|
|
@pytest.fixture()
|
|
def session_factory() -> Callable[[], Session]:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False, future=True)
|
|
|
|
def _factory() -> Session:
|
|
return factory()
|
|
|
|
return _factory
|
|
|
|
|
|
@pytest.fixture()
|
|
def unit_of_work_factory(session_factory: Callable[[], Session]) -> Callable[[], UnitOfWork]:
|
|
def _factory() -> UnitOfWork:
|
|
return UnitOfWork(session_factory=session_factory)
|
|
|
|
return _factory
|
|
|
|
|
|
def _settings(**overrides: Any) -> AdminBootstrapSettings:
|
|
defaults: dict[str, Any] = {
|
|
"email": "admin@example.com",
|
|
"username": "admin",
|
|
"password": "changeme",
|
|
"roles": ("admin", "viewer"),
|
|
"force_reset": False,
|
|
}
|
|
defaults.update(overrides)
|
|
return AdminBootstrapSettings(
|
|
email=str(defaults["email"]),
|
|
username=str(defaults["username"]),
|
|
password=str(defaults["password"]),
|
|
roles=tuple(defaults["roles"]),
|
|
force_reset=bool(defaults["force_reset"]),
|
|
)
|
|
|
|
|
|
def test_bootstrap_creates_admin_and_roles(unit_of_work_factory: Callable[[], UnitOfWork]) -> None:
|
|
settings = _settings()
|
|
|
|
role_result, admin_result = bootstrap_admin(
|
|
settings=settings,
|
|
unit_of_work_factory=unit_of_work_factory,
|
|
)
|
|
|
|
assert role_result == RoleBootstrapResult(created=4, ensured=4)
|
|
assert admin_result == AdminBootstrapResult(
|
|
created_user=True,
|
|
updated_user=False,
|
|
password_rotated=False,
|
|
roles_granted=2,
|
|
)
|
|
|
|
with unit_of_work_factory() as uow:
|
|
users_repo = uow.users
|
|
assert users_repo is not None
|
|
user = users_repo.get_by_email(settings.email, with_roles=True)
|
|
assert user is not None
|
|
assert user.is_superuser is True
|
|
assert {role.name for role in user.roles} == {"admin", "viewer"}
|
|
|
|
|
|
def test_bootstrap_is_idempotent(unit_of_work_factory: Callable[[], UnitOfWork]) -> None:
|
|
settings = _settings()
|
|
|
|
bootstrap_admin(settings=settings,
|
|
unit_of_work_factory=unit_of_work_factory)
|
|
role_result, admin_result = bootstrap_admin(
|
|
settings=settings,
|
|
unit_of_work_factory=unit_of_work_factory,
|
|
)
|
|
|
|
assert role_result.created == 0
|
|
assert role_result.ensured == 4
|
|
assert admin_result.created_user is False
|
|
assert admin_result.updated_user is False
|
|
assert admin_result.roles_granted == 0
|
|
|
|
|
|
def test_bootstrap_respects_force_reset(unit_of_work_factory: Callable[[], UnitOfWork]) -> None:
|
|
base_settings = _settings(password="initial")
|
|
bootstrap_admin(settings=base_settings,
|
|
unit_of_work_factory=unit_of_work_factory)
|
|
|
|
rotated_settings = _settings(password="rotated", force_reset=True)
|
|
_, admin_result = bootstrap_admin(
|
|
settings=rotated_settings,
|
|
unit_of_work_factory=unit_of_work_factory,
|
|
)
|
|
|
|
assert admin_result.password_rotated is True
|
|
assert admin_result.updated_user is True
|
|
|
|
with unit_of_work_factory() as uow:
|
|
users_repo = uow.users
|
|
assert users_repo is not None
|
|
user = users_repo.get_by_email(rotated_settings.email)
|
|
assert user is not None
|
|
assert user.verify_password("rotated")
|