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 models import MiningOperationType, Project from services.bootstrap import ( AdminBootstrapResult, PricingBootstrapResult, RoleBootstrapResult, bootstrap_admin, bootstrap_pricing_settings, ) from services.pricing import PricingMetadata 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") def test_bootstrap_pricing_creates_defaults(unit_of_work_factory: Callable[[], UnitOfWork]) -> None: metadata = PricingMetadata( default_payable_pct=95.0, default_currency="CAD", moisture_threshold_pct=3.0, moisture_penalty_per_pct=1.25, ) result = bootstrap_pricing_settings( metadata=metadata, unit_of_work_factory=unit_of_work_factory, ) assert isinstance(result, PricingBootstrapResult) assert result.seed.created is True assert result.projects_assigned == 0 with unit_of_work_factory() as uow: settings_repo = uow.pricing_settings assert settings_repo is not None stored = settings_repo.get_by_slug("default") assert stored.default_currency == "CAD" assert float(stored.default_payable_pct) == pytest.approx(95.0) assert float(stored.moisture_threshold_pct) == pytest.approx(3.0) assert float(stored.moisture_penalty_per_pct) == pytest.approx(1.25) def test_bootstrap_pricing_assigns_projects(unit_of_work_factory: Callable[[], UnitOfWork]) -> None: metadata = PricingMetadata( default_payable_pct=90.0, default_currency="USD", moisture_threshold_pct=5.0, moisture_penalty_per_pct=0.5, ) with unit_of_work_factory() as uow: projects_repo = uow.projects assert projects_repo is not None project = Project( name="Project Alpha", operation_type=MiningOperationType.OPEN_PIT, ) created = projects_repo.create(project) project_id = created.id result = bootstrap_pricing_settings( metadata=metadata, unit_of_work_factory=unit_of_work_factory, ) assert result.projects_assigned == 1 assert result.seed.created is True with unit_of_work_factory() as uow: projects_repo = uow.projects assert projects_repo is not None stored = projects_repo.get(project_id, with_pricing=True) assert stored.pricing_settings is not None assert stored.pricing_settings.default_currency == "USD" def test_bootstrap_pricing_is_idempotent(unit_of_work_factory: Callable[[], UnitOfWork]) -> None: metadata = PricingMetadata( default_payable_pct=92.5, default_currency="EUR", moisture_threshold_pct=4.5, moisture_penalty_per_pct=0.75, ) first = bootstrap_pricing_settings( metadata=metadata, unit_of_work_factory=unit_of_work_factory, ) second = bootstrap_pricing_settings( metadata=metadata, unit_of_work_factory=unit_of_work_factory, ) assert first.seed.created is True assert second.seed.created is False assert second.projects_assigned == 0