X-Financial/server/tests/test_auth_service.py

from __future__ import annotations

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.db.base import Base
from app.models.auth_session import AuthSession
from app.models.user_session_metric import UserSessionMetric
from app.schemas.auth import LoginRequest
from app.schemas.settings import SettingsWrite
from app.services.auth import AuthenticatedUser, AuthService
from app.services.auth_sessions import AuthSessionService
from app.services.employee import EmployeeService
from app.services.settings import SettingsService


def build_session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    return session_factory()


def test_employee_can_login_with_seed_default_password() -> None:
    with build_session() as db:
        employee = EmployeeService(db).list_employees()[0]
        result = AuthService(db).login(
            LoginRequest(username=employee.email, password="123456")
        )

        assert result.ok is True
        assert result.user.username == employee.email
        assert result.user.name == employee.name
        assert result.user.position == employee.position
        assert result.user.grade == employee.grade
        assert result.user.roleCodes
        assert result.user.isAdmin is False
        assert result.accessToken
        assert result.tokenType == "Bearer"
        stored_session = db.query(AuthSession).one()
        assert stored_session.token_hash == AuthSessionService.hash_token(result.accessToken)
        assert stored_session.token_hash != result.accessToken


def test_current_user_snapshot_refreshes_employee_position() -> None:
    with build_session() as db:
        employee = EmployeeService(db).list_employees()[0]
        result = AuthService(db).get_user_snapshot(employee.email)

        assert result is not None
        assert result.username == employee.email
        assert result.name == employee.name
        assert result.department == employee.department
        assert result.position == employee.position
        assert result.grade == employee.grade


def test_admin_can_login_with_database_password() -> None:
    with build_session() as db:
        settings_service = SettingsService(db)
        payload = settings_service.get_settings_snapshot().model_dump()
        payload["adminForm"]["adminAccount"] = "superadmin"
        payload["adminForm"]["newPassword"] = "admin123"
        payload["adminForm"]["confirmPassword"] = "admin123"
        settings_service.save_settings_snapshot(SettingsWrite(**payload))

        result = AuthService(db).login(
            LoginRequest(username="superadmin", password="admin123")
        )

        assert result.ok is True
        assert result.user.username == "superadmin"
        assert result.user.isAdmin is True
        assert result.user.position == "系统管理员"
        assert result.user.roleCodes == ["manager"]


def test_disabled_employee_cannot_login() -> None:
    with build_session() as db:
        service = EmployeeService(db)
        employee = service.list_employees()[0]
        service.disable_employee(employee.id)

        try:
            AuthService(db).login(LoginRequest(username=employee.email, password="123456"))
        except ValueError as exc:
            assert "账号或密码错误" in str(exc)
        else:
            raise AssertionError("disabled employee login should be rejected")


def test_reenabled_employee_can_login_again() -> None:
    with build_session() as db:
        service = EmployeeService(db)
        employee = service.list_employees()[0]
        service.disable_employee(employee.id)
        service.enable_employee(employee.id)

        result = AuthService(db).login(LoginRequest(username=employee.email, password="123456"))

        assert result.ok is True
        assert result.user.username == employee.email


def test_employee_login_skips_directory_bootstrap_when_employee_exists(monkeypatch) -> None:
    with build_session() as db:
        service = AuthService(db)
        calls: list[str] = []

        class ExistingEmployee:
            email = "demo@example.com"
            password_hash = "hash"
            employment_status = "在职"

        def fail_if_bootstrapped(self) -> None:
            calls.append("ensure_directory_ready")
            raise AssertionError("existing employee login should not run directory bootstrap")

        monkeypatch.setattr(
            AuthService,
            "_find_employee_by_email",
            lambda self, _: ExistingEmployee(),
        )
        monkeypatch.setattr(
            "app.services.auth.verify_password",
            lambda password, password_hash: True,
        )
        monkeypatch.setattr(
            AuthService,
            "_build_employee_user",
            lambda self, employee: AuthenticatedUser(
                username=employee.email,
                name="Demo",
                role="使用者",
                department="",
                position="",
                grade="",
                employee_no="",
                manager_name="",
                location="",
                cost_center="",
                finance_owner_name="",
                risk_profile={},
                role_codes=["user"],
                email=employee.email,
                avatar="D",
            ),
        )
        monkeypatch.setattr(EmployeeService, "ensure_directory_ready", fail_if_bootstrapped)

        user = service._authenticate_employee("demo@example.com", "123456")

        assert user is not None
        assert user.username == "demo@example.com"
        assert calls == []


def test_login_session_write_rolls_back_metric_when_token_issue_fails(monkeypatch) -> None:
    with build_session() as db:
        user = AuthenticatedUser(
            username="rollback@example.com",
            name="Rollback User",
            role="使用者",
            department="",
            position="",
            grade="",
            employee_no="",
            manager_name="",
            location="",
            cost_center="",
            finance_owner_name="",
            risk_profile={},
            role_codes=["user"],
            email="rollback@example.com",
            avatar="R",
        )

        def fail_issue(*args, **kwargs):
            raise RuntimeError("token issue failed")

        monkeypatch.setattr(AuthSessionService, "issue", fail_issue)

        with pytest.raises(RuntimeError, match="token issue failed"):
            AuthService(db)._build_login_response(user)

        assert db.query(AuthSession).count() == 0
        assert db.query(UserSessionMetric).count() == 0