company-ai-platform/tests/test_feishu_users.py

from collections.abc import Iterator

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.core.config import get_settings
from app.core.database import Base, get_db
from app.modules.audit.models import AuditLog
from app.modules.business.models import MarketWatchlist
from app.modules.feishu_users.constants import (
    FeishuCapability,
    FeishuUserAuditAction,
    FeishuUserRole,
    FeishuUserStatus,
    parse_admin_identities,
)
from app.modules.feishu_users.models import (
    FeishuAdminBootstrapTombstone,
    FeishuUser,
)
from app.modules.feishu_users.routes import router
from app.modules.feishu_users.services import (
    FeishuIdentityService,
    FeishuUserManagementService,
)


@pytest.fixture
def session_factory() -> Iterator[sessionmaker[Session]]:
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(
        engine,
        tables=[
            AuditLog.__table__,
            FeishuAdminBootstrapTombstone.__table__,
            FeishuUser.__table__,
            MarketWatchlist.__table__,
        ],
    )
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    try:
        yield factory
    finally:
        Base.metadata.drop_all(
            engine,
            tables=[
                MarketWatchlist.__table__,
                FeishuAdminBootstrapTombstone.__table__,
                FeishuUser.__table__,
                AuditLog.__table__,
            ],
        )
        engine.dispose()


def test_identity_registration_is_tenant_scoped_and_bootstraps_admin(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        service = FeishuIdentityService(
            db,
            admin_identities="tenant-a:ou-admin",
        )
        admin = service.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-admin",
            union_id="on-admin",
            user_id="u-admin",
        )
        ordinary = service.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        other_tenant = service.resolve_or_register(
            tenant_key="tenant-b",
            open_id="ou-user",
        )

        assert admin.role == FeishuUserRole.ADMIN
        assert admin.is_admin is True
        assert admin.has_capability(FeishuCapability.USER_ADMINISTRATION) is True
        assert ordinary.role == FeishuUserRole.USER
        assert ordinary.has_capability(FeishuCapability.PERSONAL_AI) is True
        assert ordinary.has_capability(FeishuCapability.COMPANY_REPORTS) is False
        assert ordinary.owner_id != other_tenant.owner_id
        assert db.scalar(select(FeishuUser).where(FeishuUser.id == admin.owner_id)).union_id == (
            "on-admin"
        )

        registrations = list(
            db.execute(
                select(AuditLog).where(
                    AuditLog.action == FeishuUserAuditAction.REGISTER
                )
            ).scalars()
        )
        assert len(registrations) == 3
        assert all("ou-" not in (item.request_payload or "") for item in registrations)


def test_each_configured_admin_identity_can_bootstrap_once(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        service = FeishuIdentityService(
            db,
            admin_identities="tenant-a:ou-admin-a,tenant-b:ou-admin-b",
        )

        first = service.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-admin-a",
        )
        second = service.resolve_or_register(
            tenant_key="tenant-b",
            open_id="ou-admin-b",
        )

        assert first.role == FeishuUserRole.ADMIN
        assert second.role == FeishuUserRole.ADMIN


def test_identity_refreshes_existing_sender_without_reapplying_admin_bootstrap(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        principal = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-existing",
        )
        refreshed = FeishuIdentityService(
            db,
            admin_identities="tenant-a:ou-existing",
        ).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-existing",
            union_id="on-existing",
        )

        assert refreshed.owner_id == principal.owner_id
        assert refreshed.role == FeishuUserRole.USER
        assert refreshed.union_id == "on-existing"
        assert db.scalar(
            select(AuditLog).where(
                AuditLog.action == FeishuUserAuditAction.AUTHENTICATE
            )
        )


def test_first_registration_claims_only_matching_legacy_watchlist(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        db.add_all(
            [
                MarketWatchlist(
                    actor="ou-claim",
                    symbol="600000.SH",
                    enabled=True,
                ),
                MarketWatchlist(
                    actor="ou-other",
                    symbol="000001.SZ",
                    enabled=True,
                ),
            ]
        )
        db.commit()

        principal = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-claim",
            open_id="ou-claim",
        )

        records = list(
            db.execute(
                select(MarketWatchlist).order_by(MarketWatchlist.id.asc())
            ).scalars()
        )
        assert records[0].owner_id == principal.owner_id
        assert records[1].owner_id is None


def test_identity_rejects_missing_parts_and_invalid_admin_config(
    session_factory: sessionmaker[Session],
) -> None:
    assert parse_admin_identities("tenant-a:ou-a, tenant-b:ou-b") == {
        ("tenant-a", "ou-a"),
        ("tenant-b", "ou-b"),
    }
    with pytest.raises(ValueError):
        parse_admin_identities("missing-separator")

    with session_factory() as db:
        with pytest.raises(HTTPException) as exc_info:
            FeishuIdentityService(db).resolve_or_register(
                tenant_key="tenant-a",
                open_id=" ",
            )
        assert exc_info.value.status_code == 422
        assert db.scalar(select(FeishuUser)) is None


def test_management_protects_last_active_admin_and_audits_denial(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        identity = FeishuIdentityService(
            db,
            admin_identities="tenant-a:ou-admin",
        )
        admin = identity.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-admin",
        )
        user = identity.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        management = FeishuUserManagementService(db)

        with pytest.raises(HTTPException) as demote_error:
            management.update_user(
                admin.user_code,
                changes={"role": FeishuUserRole.USER},
                actor="service-admin",
            )
        assert demote_error.value.status_code == 409

        with pytest.raises(HTTPException) as disable_error:
            management.update_user(
                admin.user_code,
                changes={"status": FeishuUserStatus.DISABLED},
                actor="service-admin",
            )
        assert disable_error.value.status_code == 409

        promoted = management.update_user(
            user.user_code,
            changes={"role": FeishuUserRole.ADMIN},
            actor="service-admin",
        )
        assert promoted.role == FeishuUserRole.ADMIN
        demoted = management.update_user(
            admin.user_code,
            changes={"role": FeishuUserRole.USER},
            actor="service-admin",
        )
        assert demoted.role == FeishuUserRole.USER

        denied = list(
            db.execute(
                select(AuditLog).where(
                    AuditLog.action == FeishuUserAuditAction.UPDATE_DENIED
                )
            ).scalars()
        )
        assert len(denied) == 2
        assert all(item.actor == "service-admin" for item in denied)


def test_management_validates_timezone_and_quiet_hours(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        user = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        management = FeishuUserManagementService(db)
        updated = management.update_user(
            user.user_code,
            changes={
                "timezone": "Europe/London",
                "quiet_hours_start": "22:30",
                "quiet_hours_end": "07:15",
            },
            actor="service-admin",
        )
        assert updated.timezone == "Europe/London"
        assert updated.quiet_hours_start.isoformat() == "22:30:00"
        assert updated.quiet_hours_end.isoformat() == "07:15:00"

        with pytest.raises(HTTPException) as timezone_error:
            management.update_user(
                user.user_code,
                changes={"timezone": "Invalid/Nowhere"},
                actor="service-admin",
            )
        assert timezone_error.value.status_code == 422

        with pytest.raises(HTTPException) as quiet_error:
            management.update_user(
                user.user_code,
                changes={"quiet_hours_end": None},
                actor="service-admin",
            )
        assert quiet_error.value.status_code == 422

        cleared = management.update_user(
            user.user_code,
            changes={
                "quiet_hours_start": None,
                "quiet_hours_end": None,
            },
            actor="service-admin",
        )
        assert cleared.quiet_hours_start is None
        assert cleared.quiet_hours_end is None


def test_disabled_principal_has_no_capabilities(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        principal = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        FeishuUserManagementService(db).update_user(
            principal.user_code,
            changes={"status": FeishuUserStatus.DISABLED},
            actor="service-admin",
        )
        refreshed = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        assert refreshed.is_active is False
        assert refreshed.has_capability(FeishuCapability.PERSONAL_AI) is False
        with pytest.raises(HTTPException) as exc_info:
            refreshed.require_capability(FeishuCapability.PERSONAL_AI)
        assert exc_info.value.status_code == 403


def test_internal_user_routes_use_api_principal_and_ignore_body_identity(
    session_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("API_KEY", "service-key")
    monkeypatch.setenv("API_ACTOR", "service-admin")
    monkeypatch.setenv("API_KEYS", "[]")
    get_settings.cache_clear()

    with session_factory() as db:
        identity = FeishuIdentityService(
            db,
            admin_identities="tenant-a:ou-admin",
        )
        identity.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-admin",
        )
        user = identity.resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-user",
        )
        user_code = user.user_code

    app = FastAPI()
    app.include_router(router, prefix="/api/v1/integrations/feishu")

    def override_db() -> Iterator[Session]:
        with session_factory() as db:
            yield db

    app.dependency_overrides[get_db] = override_db
    client = TestClient(app)
    try:
        assert (
            client.get("/api/v1/integrations/feishu/users").status_code
            == 401
        )
        response = client.patch(
            f"/api/v1/integrations/feishu/users/{user_code}",
            headers={"X-API-Key": "service-key"},
            json={
                "role": "admin",
                "actor": "forged-user",
                "open_id": "ou-forged",
            },
        )
        assert response.status_code == 200
        assert response.json()["role"] == "admin"
        assert response.json()["open_id"] == "ou-user"

        listed = client.get(
            "/api/v1/integrations/feishu/users?role=admin",
            headers={"X-API-Key": "service-key"},
        )
        assert listed.status_code == 200
        assert listed.json()["total"] == 2
        detail = client.get(
            f"/api/v1/integrations/feishu/users/{user_code}",
            headers={"X-API-Key": "service-key"},
        )
        assert detail.status_code == 200

        with session_factory() as db:
            update_log = db.execute(
                select(AuditLog)
                .where(AuditLog.action == FeishuUserAuditAction.UPDATE)
                .order_by(AuditLog.id.desc())
            ).scalars().first()
            assert update_log is not None
            assert update_log.actor == "service-admin"
            assert "forged-user" not in (update_log.request_payload or "")
            assert "ou-forged" not in (update_log.request_payload or "")
    finally:
        get_settings.cache_clear()