hunjing/backend/tests/test_models.py

"""首批模型的集成测试：CRUD、约束与外键行为（运行于 PostgreSQL）。"""

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.core.ids import new_id
from app.models import AuditEvent, OutboxEvent, PersonalDomain, ServiceAccount, Session, User


async def _make_user(db: async_sessionmaker[AsyncSession], email: str = "a@example.com") -> User:
    async with db() as session:
        user = User(email=email, display_name="测试用户")
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user


async def test_user_crud_and_id_prefix(db: async_sessionmaker[AsyncSession]) -> None:
    user = await _make_user(db)
    assert user.id.startswith("user_")
    async with db() as session:
        loaded = await session.get(User, user.id)
        assert loaded is not None
        assert loaded.email == "a@example.com"
        assert loaded.created_at.tzinfo is not None  # aware UTC


async def test_duplicate_email_rejected_by_unique(db: async_sessionmaker[AsyncSession]) -> None:
    await _make_user(db)
    async with db() as session:
        session.add(User(email="a@example.com", display_name="重复"))
        with pytest.raises(IntegrityError):
            await session.commit()


async def test_personal_domain_requires_user(db: async_sessionmaker[AsyncSession]) -> None:
    async with db() as session:
        session.add(PersonalDomain(user_id=new_id("user_"), name="镜域"))
        with pytest.raises(IntegrityError):
            await session.commit()


async def test_delete_user_restricted_by_domain(db: async_sessionmaker[AsyncSession]) -> None:
    user = await _make_user(db)
    async with db() as session:
        session.add(PersonalDomain(user_id=user.id, name="镜域"))
        await session.commit()
    async with db() as session:
        loaded = await session.get(User, user.id)
        assert loaded is not None
        await session.delete(loaded)
        with pytest.raises(IntegrityError):
            await session.commit()


async def test_session_cascades_with_user(db: async_sessionmaker[AsyncSession]) -> None:
    user = await _make_user(db)
    async with db() as session:
        session.add(
            Session(
                id="opaque-session-token",
                user_id=user.id,
                token_hash="sha256:opaque-session-token",
                csrf_token_hash="sha256:csrf",
                expires_at=datetime.now(UTC) + timedelta(days=7),
            )
        )
        await session.commit()
    async with db() as session:
        loaded = await session.get(User, user.id)
        assert loaded is not None
        await session.delete(loaded)
        await session.commit()
    async with db() as session:
        assert (await session.scalars(select(Session))).all() == []


async def test_service_account_unique_domain_name(db: async_sessionmaker[AsyncSession]) -> None:
    user = await _make_user(db)
    async with db() as session:
        domain = PersonalDomain(user_id=user.id, name="镜域")
        session.add(domain)
        await session.commit()
        await session.refresh(domain)
    async with db() as session:
        session.add(
            ServiceAccount(
                domain_id=domain.id, name="ci", key_prefix="hjk_ci", key_hash="sha256:..."
            )
        )
        await session.commit()
        session.add(
            ServiceAccount(
                domain_id=domain.id, name="ci", key_prefix="hjk_ci2", key_hash="sha256:..."
            )
        )
        with pytest.raises(IntegrityError):
            await session.commit()


async def test_outbox_status_check_constraint(db: async_sessionmaker[AsyncSession]) -> None:
    async with db() as session:
        session.add(
            OutboxEvent(
                aggregate_type="user",
                aggregate_id=new_id("user_"),
                event_type="user.created",
                payload={},
                status="bogus",
            )
        )
        with pytest.raises(IntegrityError):
            await session.commit()


async def test_audit_and_outbox_same_transaction(db: async_sessionmaker[AsyncSession]) -> None:
    """契约第 4.4 节：权威状态、AuditEvent 与 OutboxEvent 同事务提交。"""
    async with db() as session:
        user = User(email="tx@example.com", display_name="事务")
        session.add(user)
        await session.flush()  # 取得 user.id，事务尚未提交
        session.add(
            AuditEvent(
                actor_type="system",
                actor_id="system",
                action="user.created",
                aggregate_type="user",
                aggregate_id=user.id,
            )
        )
        session.add(
            OutboxEvent(
                aggregate_type="user",
                aggregate_id=user.id,
                event_type="user.created",
                payload={"user_id": user.id},
            )
        )
        await session.commit()
        assert (await session.scalars(select(OutboxEvent))).one().status == "pending"