269 lines
9.2 KiB
Python
269 lines
9.2 KiB
Python
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.tenant import Tenant
|
|
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 configure_admin(db: Session, *, account: str = "admin", password: str = "admin") -> None:
|
|
settings_service = SettingsService(db)
|
|
payload = settings_service.get_settings_snapshot().model_dump()
|
|
payload["adminForm"]["adminAccount"] = account
|
|
payload["adminForm"]["newPassword"] = password
|
|
payload["adminForm"]["confirmPassword"] = password
|
|
settings_service.save_settings_snapshot(SettingsWrite(**payload))
|
|
|
|
|
|
def test_employee_can_login_with_seed_default_password() -> None:
|
|
with build_session() as db:
|
|
employee = EmployeeService(db, tenant_id="default").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, tenant_id="default").list_employees()[0]
|
|
result = AuthService(db).get_user_snapshot(
|
|
employee.email,
|
|
tenant_id="default",
|
|
)
|
|
|
|
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:
|
|
configure_admin(db, account="superadmin", password="admin123")
|
|
|
|
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"]
|
|
assert result.user.tenantId == "platform"
|
|
|
|
|
|
def test_admin_login_uses_requested_active_tenant_and_restores_session_scope() -> None:
|
|
with build_session() as db:
|
|
configure_admin(db)
|
|
db.add(
|
|
Tenant(
|
|
tenant_id="default",
|
|
tenant_code="default",
|
|
name="默认企业",
|
|
status="active",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
result = AuthService(db).login(
|
|
LoginRequest(username="admin", password="admin", tenantId="default")
|
|
)
|
|
auth_session = db.query(AuthSession).one()
|
|
restored = AuthService(db).get_session_user(auth_session)
|
|
|
|
assert result.user.isAdmin is True
|
|
assert result.user.tenantId == "default"
|
|
assert auth_session.principal_type == "admin"
|
|
assert auth_session.tenant_id == "default"
|
|
assert restored is not None
|
|
assert restored.is_admin is True
|
|
assert restored.tenant_id == "default"
|
|
|
|
|
|
def test_admin_login_rejects_unknown_or_inactive_requested_tenant() -> None:
|
|
with build_session() as db:
|
|
configure_admin(db)
|
|
db.add(
|
|
Tenant(
|
|
tenant_id="tenant-disabled",
|
|
tenant_code="company-disabled",
|
|
name="停用企业",
|
|
status="disabled",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match="企业代码不存在或当前不可用"):
|
|
AuthService(db).login(
|
|
LoginRequest(
|
|
username="admin",
|
|
password="admin",
|
|
tenantId="company-disabled",
|
|
)
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="企业代码不存在或当前不可用"):
|
|
AuthService(db).login(
|
|
LoginRequest(
|
|
username="admin",
|
|
password="admin",
|
|
tenantId="missing-tenant",
|
|
)
|
|
)
|
|
|
|
|
|
def test_disabled_employee_cannot_login() -> None:
|
|
with build_session() as db:
|
|
service = EmployeeService(db, tenant_id="default")
|
|
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, tenant_id="default")
|
|
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, _, requested_tenant: 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",
|
|
tenant_id="default",
|
|
),
|
|
)
|
|
monkeypatch.setattr(EmployeeService, "ensure_directory_ready", fail_if_bootstrapped)
|
|
|
|
user = service._authenticate_employee(
|
|
"demo@example.com",
|
|
"123456",
|
|
requested_tenant="default",
|
|
)
|
|
|
|
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",
|
|
tenant_id="default",
|
|
)
|
|
|
|
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
|