fix: create a personal cost center also on first LDAP login

This commit is contained in:
behrinml
2026-06-25 20:10:25 +02:00
parent a459bf45d1
commit 15d6e82b8c
2 changed files with 100 additions and 0 deletions
+6
View File
@@ -67,6 +67,7 @@ from backend.app.services.email_service import (
save_smtp_settings,
send_email,
)
from backend.app.services.finance_defaults import ensure_user_finance_defaults
_logger = logging.getLogger(__name__)
@@ -409,6 +410,9 @@ async def login(raw_request: Request, request: LoginRequest, response: Response,
if user and ldap_user:
# Update email and group mappings on each login
await _sync_ldap_user(db, user, ldap_user, ldap_config)
# Keep finance defaults idempotently in sync for LDAP users
# (wallet + private cost center + self-membership).
await ensure_user_finance_defaults(db, user)
except Exception as e: # SEC-AUTH-EXC: LDAP failure sets ldap_user=None, downstream local-auth path runs with its own credential check (no implicit grant)
import logging
@@ -1205,6 +1209,8 @@ async def _provision_ldap_user(db: AsyncSession, ldap_user, ldap_config) -> User
new_user.groups = list(groups_result.scalars().all())
db.add(new_user)
await db.flush()
await ensure_user_finance_defaults(db, new_user)
await db.commit()
await db.refresh(new_user)
logger.info("Auto-provisioned LDAP user: %s (groups: %s)", new_user.username, mapped_group_names)
@@ -20,8 +20,10 @@ from unittest.mock import patch
import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.models.finance import CostCenter, CostCenterMember, UserWallet
from backend.app.models.settings import Settings
from backend.app.models.user import User
from backend.app.services.ldap_service import LDAPSearchResult, LDAPUserInfo
@@ -367,3 +369,95 @@ class TestLdapProvisionRoute:
body = response.json()
group_names = {g["name"] for g in body["groups"]}
assert "Operators" in group_names
class TestLdapLoginFinanceDefaults:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_successful_ldap_login_backfills_finance_defaults(
self, async_client: AsyncClient, db_session: AsyncSession
):
"""LDAP login should ensure wallet + private cost center defaults exist.
Regression: LDAP users created before finance defaults were introduced can
exist without wallet/private center. A successful LDAP login must backfill
these defaults so billing-enabled flows have a valid personal cost center.
"""
await async_client.post(
"/api/v1/auth/setup",
json={
"auth_enabled": True,
"admin_username": "ldapadmin",
"admin_password": "AdminPass1!",
},
)
await _seed_ldap_settings(db_session, ldap_auto_provision="false")
legacy_user = User(
username="legacyldap",
email="legacyldap@test.com",
password_hash=None,
role="user",
auth_source="ldap",
is_active=True,
)
db_session.add(legacy_user)
await db_session.commit()
await db_session.refresh(legacy_user)
# Precondition: legacy LDAP row has no finance defaults yet.
wallet_before = (
await db_session.execute(select(UserWallet).where(UserWallet.user_id == legacy_user.id))
).scalar_one_or_none()
private_cc_before = (
await db_session.execute(
select(CostCenter).where(
CostCenter.owner_user_id == legacy_user.id,
CostCenter.is_private.is_(True),
)
)
).scalar_one_or_none()
assert wallet_before is None
assert private_cc_before is None
fake_ldap = LDAPUserInfo(
username="legacyldap",
email="legacyldap@test.com",
display_name="Legacy LDAP",
groups=[],
)
with patch("backend.app.services.ldap_service.authenticate_ldap_user", return_value=fake_ldap):
response = await async_client.post(
"/api/v1/auth/login",
json={"username": "legacyldap", "password": "irrelevant"},
)
assert response.status_code == 200
assert response.json()["user"]["auth_source"] == "ldap"
wallet_after = (
await db_session.execute(select(UserWallet).where(UserWallet.user_id == legacy_user.id))
).scalar_one_or_none()
assert wallet_after is not None
private_cc_after = (
await db_session.execute(
select(CostCenter).where(
CostCenter.owner_user_id == legacy_user.id,
CostCenter.is_private.is_(True),
)
)
).scalar_one_or_none()
assert private_cc_after is not None
assert private_cc_after.name == "legacyldap"
membership = (
await db_session.execute(
select(CostCenterMember).where(
CostCenterMember.cost_center_id == private_cc_after.id,
CostCenterMember.user_id == legacy_user.id,
)
)
).scalar_one_or_none()
assert membership is not None
assert membership.can_print is True