90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core import service_account as service_account_module
|
|
from app.core.service_account import (
|
|
SERVICE_ACCOUNT_EMAIL,
|
|
SYSTEM_ACCOUNT_DISPLAY_CODE,
|
|
ensure_service_account,
|
|
)
|
|
from app.models.account import Account
|
|
from app.models.user import User
|
|
|
|
|
|
class _SessionFactoryOverride:
|
|
def __init__(self, session):
|
|
self._session = session
|
|
|
|
def __call__(self):
|
|
return self
|
|
|
|
async def __aenter__(self):
|
|
return self._session
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_service_account_creates_and_reuses_seeded_user(test_db, monkeypatch):
|
|
monkeypatch.setattr(
|
|
service_account_module,
|
|
"_admin_session_factory",
|
|
_SessionFactoryOverride(test_db),
|
|
)
|
|
|
|
service_account_id = await ensure_service_account(test_db)
|
|
|
|
created_user = (
|
|
await test_db.execute(select(User).where(User.id == service_account_id))
|
|
).scalar_one()
|
|
assert created_user.email == SERVICE_ACCOUNT_EMAIL
|
|
assert created_user.is_service_account is True
|
|
|
|
system_account = (
|
|
await test_db.execute(
|
|
select(Account).where(Account.display_code == SYSTEM_ACCOUNT_DISPLAY_CODE)
|
|
)
|
|
).scalar_one()
|
|
assert created_user.account_id == system_account.id
|
|
|
|
second_id = await ensure_service_account(test_db)
|
|
assert second_id == service_account_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_service_account_marks_existing_user_as_service_account(test_db, monkeypatch):
|
|
monkeypatch.setattr(
|
|
service_account_module,
|
|
"_admin_session_factory",
|
|
_SessionFactoryOverride(test_db),
|
|
)
|
|
|
|
system_account = (
|
|
await test_db.execute(
|
|
select(Account).where(Account.display_code == SYSTEM_ACCOUNT_DISPLAY_CODE)
|
|
)
|
|
).scalar_one()
|
|
|
|
existing_user = User(
|
|
email=SERVICE_ACCOUNT_EMAIL,
|
|
name="ResolutionFlow",
|
|
password_hash="!service-account-no-login",
|
|
role="engineer",
|
|
is_super_admin=False,
|
|
is_team_admin=False,
|
|
is_active=True,
|
|
is_service_account=False,
|
|
must_change_password=False,
|
|
account_id=system_account.id,
|
|
account_role="engineer",
|
|
)
|
|
test_db.add(existing_user)
|
|
await test_db.commit()
|
|
|
|
resolved_id = await ensure_service_account(test_db)
|
|
await test_db.refresh(existing_user)
|
|
|
|
assert resolved_id == existing_user.id
|
|
assert existing_user.is_service_account is True
|