yunke/yunke-server/tests/test_management_models.py

"""管理域 6 张新表的模型冒烟测试(批次A Task1)。

只验证:建表可用、主键/唯一约束生效、默认值正确。业务逻辑在
first_reply_facts / employee_metrics 的测试里。
"""
import pytest
import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError

from app.db.models import (
    EmployeeCallDaily,
    EmployeeMetricDaily,
    EmployeeScoreConfig,
    EmployeeScoreSnapshot,
    FirstReplyFact,
    ManagementFollowup,
)


async def test_first_reply_fact_roundtrip(make_session):
    s = make_session
    s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1",
                         thread_end_ts=1_700_000_000_000,
                         reply_ts=1_700_000_600_000,
                         day=20260901, fr_ms=600_000))
    await s.commit()
    val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one()
    assert val == 600_000


async def test_first_reply_fact_unique_sample(make_session):
    s = make_session
    kwargs = {"channel": "wechat", "owner_account_id": "wx1", "peer_id": "p1",
              "thread_end_ts": 1, "reply_ts": 2, "day": 20260901, "fr_ms": 1}
    s.add(FirstReplyFact(**kwargs))
    await s.commit()
    s.add(FirstReplyFact(**kwargs))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_metric_daily_versioning_coexists(make_session):
    s = make_session
    for v in ("v1", "v2"):
        s.add(EmployeeMetricDaily(day=20260901, channel="wechat",
                                  user_id="u1", metric_version=v))
    await s.commit()
    n = (await s.execute(
        sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one()
    assert n == 2


async def test_metric_daily_defaults_zero_not_none(make_session):
    s = make_session
    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
    await s.commit()
    row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one()
    assert row.msg_sent == 0 and row.cust_threads == 0 and row.fr_sum_ms == 0
    assert row.metric_version == "v1"
    assert row.data_complete is False or row.data_complete == 0  # SQLite 0/1


async def test_metric_daily_pk_duplicate_rejected(make_session):
    """同 (day, channel, user_id, metric_version) 复合主键防重。

    Task 6 delete+insert 幂等重算的安全网:版本内重复落表直接 IntegrityError。
    """
    s = make_session
    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
    await s.commit()
    s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_call_daily_roundtrip(make_session):
    s = make_session
    s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=2,
                            call_seconds=164, answered_call_count=1,
                            call_followup_count=1))
    await s.commit()
    row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one()
    assert row.call_seconds == 164


async def test_call_daily_pk_duplicate_rejected(make_session):
    s = make_session
    s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
    await s.commit()
    s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_score_config_version_unique(make_session):
    s = make_session
    s.add(EmployeeScoreConfig(version="v1.0", name="一期试算",
                              config_json={"dimensions": {}}))
    await s.commit()
    s.add(EmployeeScoreConfig(version="v1.0", name="重复",
                              config_json={}))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_score_config_config_json_not_null(make_session):
    """config_json NOT NULL——注解与 DDL 一致(修复:去掉 | None)。

    注意 JSON 列语义:Python None 经 JSON 序列化变成字符串 'null' 而非
    SQL NULL,不会违反约束;测 NOT NULL 须用 sa.null() 显式发 SQL NULL。
    """
    s = make_session
    s.add(EmployeeScoreConfig(version="v1.0", name="空配置",
                              config_json=sa.null()))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_snapshot_batch_user_unique(make_session):
    s = make_session
    common = {"snapshot_batch_id": "b1", "period_type": "week",
              "period_start": 20260901, "period_end": 20260907,
              "score_config_version": "v1.0", "metric_version": "v1",
              "data_as_of": 20260907}
    s.add(EmployeeScoreSnapshot(user_id="u1", **common))
    await s.commit()
    s.add(EmployeeScoreSnapshot(user_id="u1", **common))
    with pytest.raises(IntegrityError):
        await s.commit()


async def test_followup_defaults(make_session):
    s = make_session
    s.add(ManagementFollowup(employee_user_id="u1", source_type="manual"))
    await s.commit()
    row = (await s.execute(sa.select(ManagementFollowup))).scalar_one()
    assert row.status == "open"
    assert row.created_at is not None