yunke/yunke-server/tests/test_repos.py

"""Tests for app.storage.repos — idempotent upserts + type coercion.

These tests build an isolated in-memory SQLite async engine (same pattern as
``tests/test_models.py``) and exercise the public upsert helpers in
``app.storage.repos``. They never touch the production MySQL engine.

Coverage:

* Idempotency: same PK twice -> one row, with non-PK fields updated.
* Type coercion: bool fields (``is_send``/``is_del``/``is_block``/``is_on_job``)
  accept int / "1"-style str / bool; ``gender`` str -> int; ``None`` preserved.
* Batch dedup: 3 messages with one duplicate ``msg_id`` -> 2 rows in the table.
* Return values: ``upsert_customers`` / ``upsert_messages`` report counts.
* Embedded-staff best-effort upsert from a wechat-account record.
* Direct unit tests on the module-level ``_coerce_bool`` / ``_coerce_int``
  helpers (kept module-level precisely so they can be exercised here).
"""

from __future__ import annotations

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.db.base import Base
from app.db.models import (
    ChatGroup,
    Customer,
    CustomerAccountLink,
    Department,
    GroupMember,
    Message,
    MomentsPost,
    PhoneCall,
    Sms,
    Staff,
    WechatAccount,
    WhatsappAccount,
)
from app.storage.repos import (
    _build_column_values,
    _coerce_bool,
    _coerce_int,
    prune_customer_links,
    upsert_chat_group,
    upsert_customer,
    upsert_customers,
    upsert_department,
    upsert_messages,
    upsert_moments_posts,
    upsert_phone_call,
    upsert_sms,
    upsert_staff,
    upsert_staff_from_account,
    upsert_voice_call,
    upsert_wechat_account,
    upsert_whatsapp_account,
)

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


async def _make_engine():
    """Create an in-memory async SQLite engine with all tables created."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    return engine


# ---------------------------------------------------------------------------
# Coercion helpers (module-level, exercised directly for unit coverage)
# ---------------------------------------------------------------------------


def test_coerce_bool_handles_documented_cases():
    """bool fields accept None / bool / int / '1'-style str."""
    assert _coerce_bool(None) is None
    assert _coerce_bool(True) is True
    assert _coerce_bool(False) is False
    assert _coerce_bool(1) is True
    assert _coerce_bool(0) is False
    assert _coerce_bool("1") is True
    assert _coerce_bool("0") is False


def test_coerce_int_handles_documented_cases():
    """gender / type-style fields: str -> int; None preserved; 0 preserved."""
    assert _coerce_int(None) is None
    assert _coerce_int(2) == 2
    assert _coerce_int("2") == 2
    assert _coerce_int(0) == 0
    assert _coerce_int("0") == 0


def test_coerce_int_returns_none_for_non_numeric_string():
    """A non-numeric string (e.g. a label) -> None.

    Integer columns (``type_raw``, ``gender``, ...) cannot store a
    non-numeric value; the coercion collapses it to ``None``.
    """
    assert _coerce_int("text") is None
    assert _coerce_int("unknown") is None


# ---------------------------------------------------------------------------
# Customer upsert
# ---------------------------------------------------------------------------


async def test_upsert_customer_shared_friend_maintains_links():
    """共享好友:档案全局单行(last-wins),link 每个账号各一行(2026-09-08)。

    同一 friend_id 先后在 ownerA/ownerB 名下 upsert——customer 只有一行
    且归属=后写;customer_account_link 两行,各自带自己的备注快照与
    fromType(来自 raw_json)。
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            base = {
                "friend_id": "wxid_shared",
                "channel": "wechat",
                "friend_nickname": "共享客户",
                "phone": "13800000000",
                "send_num": 10,
                "receive_num": 4,
            }
            await upsert_customer(session, {
                **base, "owner_account_id": "wx_a",
                "friend_remark": "A的备注", "raw_json": {"fromType": 1},
            })
            await session.commit()
            await upsert_customer(session, {
                **base, "owner_account_id": "wx_b",
                "friend_remark": "B的备注", "raw_json": {"fromType": 2},
            })
            await session.commit()

            custs = (await session.execute(
                sa.select(Customer).execution_options(populate_existing=True)
            )).scalars().all()
            assert len(custs) == 1
            assert custs[0].owner_account_id == "wx_b"  # 档案 last-wins

            links = (await session.execute(
                sa.select(CustomerAccountLink).execution_options(populate_existing=True)
            )).scalars().all()
            assert {(l.owner_account_id, l.friend_remark) for l in links} == {
                ("wx_a", "A的备注"), ("wx_b", "B的备注"),
            }
            by_owner = {l.owner_account_id: l for l in links}
            assert by_owner["wx_a"].from_type == 1
            assert by_owner["wx_b"].from_type == 2
            assert by_owner["wx_a"].send_num == 10
            assert by_owner["wx_a"].phone == "13800000000"
    finally:
        await engine.dispose()


async def test_upsert_customer_without_owner_skips_link():
    """无 owner_account_id 的档案 upsert(理论路径)只写 customer,不碰 link。"""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            await upsert_customer(session, {
                "friend_id": "wxid_solo", "channel": "wechat",
            })
            await session.commit()
            n_links = (await session.execute(
                sa.select(sa.func.count()).select_from(CustomerAccountLink)
            )).scalar_one()
            assert n_links == 0
    finally:
        await engine.dispose()


async def test_prune_customer_links_removes_departed():
    """对账:保留名单外的 link 被删;空名单删光;他人名下不动。"""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            for owner, fid in [("wx_a", "f1"), ("wx_a", "f2"), ("wx_a", "f3"),
                               ("wx_b", "f2")]:
                await upsert_customer(session, {
                    "friend_id": fid, "channel": "wechat",
                    "owner_account_id": owner,
                })
            await session.commit()

            deleted = await prune_customer_links(
                session, channel="wechat", owner_account_id="wx_a",
                keep_friend_ids=["f1", "f3"],
            )
            await session.commit()
            assert deleted == 1

            left = (await session.execute(
                sa.select(CustomerAccountLink.friend_id, CustomerAccountLink.owner_account_id)
            )).all()
            assert set(left) == {("f1", "wx_a"), ("f3", "wx_a"), ("f2", "wx_b")}

            # 空名单=该账号全部删除(wx_b 的保留)
            deleted_all = await prune_customer_links(
                session, channel="wechat", owner_account_id="wx_a",
                keep_friend_ids=[],
            )
            await session.commit()
            assert deleted_all == 2
            left2 = (await session.execute(
                sa.select(CustomerAccountLink.friend_id, CustomerAccountLink.owner_account_id)
            )).all()
            assert set(left2) == {("f2", "wx_b")}
    finally:
        await engine.dispose()


async def test_upsert_customer_is_idempotent_and_updates_version():
    """Upsert same friend_id twice with different version -> 1 row, version updated."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "friend_id": "wxid_abc",
                "channel": "wechat",
                "version": "v1",
                "friend_nickname": "Alice",
            }
            rec2 = {
                "friend_id": "wxid_abc",
                "channel": "wechat",
                "version": "v2",
                "friend_nickname": "Alice",
            }

            await upsert_customer(session, rec1)
            await session.commit()
            await upsert_customer(session, rec2)
            await session.commit()

            # ``populate_existing=True`` bypasses the identity-map cache so the
            # verify query reflects the second upsert's writes (with
            # ``expire_on_commit=False`` the loaded instance is otherwise stale).
            rows = (
                await session.execute(
                    sa.select(Customer).execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].friend_id == "wxid_abc"
            assert rows[0].version == "v2"
            assert rows[0].friend_nickname == "Alice"
    finally:
        await engine.dispose()


async def test_upsert_customer_coerces_gender_and_bool():
    """gender '2' -> int 2; is_block '1' -> True."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "friend_id": "wxid_g1",
                "channel": "wechat",
                "gender": "2",
                "is_block": "1",
            }
            await upsert_customer(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Customer).where(Customer.friend_id == "wxid_g1")
                )
            ).scalar_one()
            assert fetched.gender == 2
            assert fetched.is_block is True
    finally:
        await engine.dispose()


async def test_upsert_customer_preserves_none_for_bool_and_gender():
    """Missing / None bool & gender stay None (no spurious coercion)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "friend_id": "wxid_n",
                "channel": "wechat",
                "gender": None,
                "is_block": None,
            }
            await upsert_customer(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Customer).where(Customer.friend_id == "wxid_n")
                )
            ).scalar_one()
            assert fetched.gender is None
            assert fetched.is_block is None
    finally:
        await engine.dispose()


async def test_upsert_customer_keeps_raw_json():
    """raw_json is preserved verbatim (the whole input record)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "friend_id": "wxid_r",
                "channel": "wechat",
                "friendWxNickname": "upstream-name",  # a non-column key
                "nested": {"k": "v"},
            }
            await upsert_customer(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Customer).where(Customer.friend_id == "wxid_r")
                )
            ).scalar_one()
            assert fetched.raw_json == rec
    finally:
        await engine.dispose()


async def test_upsert_customer_coerces_list_dept_route_to_joined_string():
    """A list ``dept_route`` is stored joined ("A;B"); raw_json keeps the list.

    Regression for the live-validation fix: the real yunke friends-list API
    returns ``deptRoute`` as a LIST of dept ids, but ``Customer.dept_route`` is
    a scalar String(255) column. Without the list->";"-join coercion in
    :func:`_build_column_values` the bind raises (cannot store a list in a
    String column). This test pins the coercion: the column holds the joined
    scalar, while ``raw_json`` still carries the original list verbatim so no
    information is lost.
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "friend_id": "wxid_list",
                "channel": "wechat",
                "dept_route": ["A", "B"],  # list upstream shape
            }
            await upsert_customer(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Customer).where(Customer.friend_id == "wxid_list")
                )
            ).scalar_one()
            # Stored as the joined scalar string.
            assert fetched.dept_route == "A;B"
            # The verbatim list is still recoverable from raw_json.
            assert fetched.raw_json["dept_route"] == ["A", "B"]
    finally:
        await engine.dispose()


async def test_upsert_customers_batch_returns_count():
    """Batch upsert returns the number of records processed."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            recs = [
                {"friend_id": "wxid_b1", "channel": "wechat", "version": "v1"},
                {"friend_id": "wxid_b2", "channel": "wechat", "version": "v1"},
            ]
            count = await upsert_customers(session, recs)
            await session.commit()

            assert count == 2
            rows = (await session.execute(sa.select(Customer))).scalars().all()
            assert {r.friend_id for r in rows} == {"wxid_b1", "wxid_b2"}
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# Message upsert
# ---------------------------------------------------------------------------


async def test_upsert_messages_coerces_types():
    """is_send '1' -> True; is_del 0 -> False; type label stored as string;
    type_raw coerced to int; same msg_id twice -> 1 row."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "msg_id": "m_coerce",
                "channel": "wechat",
                "is_send": "1",
                "is_del": 0,
                "type": "text",  # normalized label — stored as-is (no coercion)
                "type_raw": "1",  # raw code arrives as str -> coerced to int
                "timestamp": 1_700_000_000_000,
            }
            await upsert_messages(session, [rec])
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Message).where(Message.msg_id == "m_coerce")
                )
            ).scalar_one()
            assert fetched.is_send is True
            assert fetched.is_del is False
            assert fetched.type == "text"  # string label preserved
            assert fetched.type_raw == 1  # raw code coerced to int
            assert fetched.timestamp == 1_700_000_000_000

            # Upsert same msg_id again with different text + new raw code ->
            # row updated, not duplicated.
            rec2 = dict(rec, text="updated", type="unknown", type_raw=3)
            await upsert_messages(session, [rec2])
            await session.commit()

            # ``populate_existing=True`` bypasses the identity-map cache so we
            # observe the second upsert's values.
            rows = (
                await session.execute(
                    sa.select(Message)
                    .where(Message.msg_id == "m_coerce")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].text == "updated"
            assert rows[0].type == "unknown"
            assert rows[0].type_raw == 3
    finally:
        await engine.dispose()


async def test_upsert_messages_preserves_none_type_and_type_raw():
    """Missing / None type & type_raw stay None (no spurious coercion)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "msg_id": "m_none",
                "channel": "wechat",
                "type": None,
                "type_raw": None,
            }
            await upsert_messages(session, [rec])
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Message).where(Message.msg_id == "m_none")
                )
            ).scalar_one()
            assert fetched.type is None
            assert fetched.type_raw is None
    finally:
        await engine.dispose()


async def test_upsert_messages_dedupes_duplicate_msg_id():
    """3 messages with 1 duplicate msg_id -> 2 rows; dup row updated."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            recs = [
                {"msg_id": "d1", "channel": "wechat", "text": "a"},
                {"msg_id": "d2", "channel": "wechat", "text": "b"},
                {"msg_id": "d1", "channel": "wechat", "text": "a-updated"},
            ]
            count = await upsert_messages(session, recs)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(Message).execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 2
            assert {r.msg_id for r in rows} == {"d1", "d2"}
            # The duplicate's later text wins.
            d1 = next(r for r in rows if r.msg_id == "d1")
            assert d1.text == "a-updated"
            # Return value reflects records processed (input count).
            assert count == 3
    finally:
        await engine.dispose()


async def test_upsert_messages_empty_list_returns_zero():
    """An empty batch is a no-op and reports 0."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            count = await upsert_messages(session, [])
            assert count == 0
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# WechatAccount + embedded Staff upsert (best-effort)
# ---------------------------------------------------------------------------


async def test_upsert_wechat_account_coerces_is_on_job():
    """WechatAccount upsert stores wx_id PK and coerces is_on_job."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "wx_id": "wxid_staff",
                "nick_name": "Alice",
                "user_id": "u1",
                "user_name": "Alice Wang",
                "is_on_job": 1,
                "contacts_num": 100,
            }
            await upsert_wechat_account(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(WechatAccount).where(
                        WechatAccount.wx_id == "wxid_staff"
                    )
                )
            ).scalar_one()
            assert fetched.nick_name == "Alice"
            assert fetched.is_on_job is True
            assert fetched.contacts_num == 100
    finally:
        await engine.dispose()


async def test_upsert_staff_from_account_extracts_user_id_and_name():
    """Embedded staff (user_id + user_name) upsert from an account record."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "wx_id": "wxid_staff",
                "user_id": "u1",
                "user_name": "Alice Wang",
                "is_on_job": 1,
            }
            await upsert_staff_from_account(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "u1")
                )
            ).scalar_one()
            assert fetched.name == "Alice Wang"
            assert fetched.is_on_job is True
    finally:
        await engine.dispose()


async def test_upsert_staff_from_account_skips_when_no_user_id():
    """No user_id -> nothing inserted (best-effort, no raise)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {"wx_id": "wxid_no_user", "nick_name": "anon"}
            # Must not raise.
            await upsert_staff_from_account(session, rec)
            await session.commit()

            rows = (await session.execute(sa.select(Staff))).scalars().all()
            assert rows == []
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# Department upsert (Task 2.0)
# ---------------------------------------------------------------------------


async def test_upsert_department_is_idempotent_and_updates_path():
    """Same dept_id twice with different path -> 1 row, path refreshed."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "dept_id": "1.1",
                "parent_id": "1",
                "name": "East",
                "path": "Sales/East",
            }
            rec2 = {
                "dept_id": "1.1",
                "parent_id": "1",
                "name": "East",
                # Re-scrape reconstructs the same path; verify it is overwritten
                # rather than producing a duplicate row.
                "path": "Sales/East",
                "raw_json": {"id": "1.1", "pId": "1", "name": "East"},
            }

            await upsert_department(session, rec1)
            await session.commit()
            await upsert_department(session, rec2)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(Department)
                    .where(Department.dept_id == "1.1")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].dept_id == "1.1"
            assert rows[0].parent_id == "1"
            assert rows[0].name == "East"
            assert rows[0].path == "Sales/East"
            # raw_json from the second upsert wins (last write).
            assert rows[0].raw_json == rec2["raw_json"]
    finally:
        await engine.dispose()


async def test_upsert_department_preserves_none_parent_for_root():
    """A root dept (parent_id None) is stored as-is; no spurious coercion."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "dept_id": "1",
                "parent_id": None,
                "name": "Sales",
                "path": "Sales",
            }
            await upsert_department(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Department).where(Department.dept_id == "1")
                )
            ).scalar_one()
            assert fetched.parent_id is None
            assert fetched.name == "Sales"
            assert fetched.path == "Sales"
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# upsert_staff — direct staff-shaped dict upsert (Task 2.1)
# ---------------------------------------------------------------------------


async def test_upsert_staff_is_idempotent_and_updates_name():
    """Same user_id twice with different name -> 1 row, name refreshed."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {"user_id": "u_new", "name": "Old Name"}
            rec2 = {"user_id": "u_new", "name": "New Name"}

            await upsert_staff(session, rec1)
            await session.commit()
            await upsert_staff(session, rec2)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(Staff)
                    .where(Staff.user_id == "u_new")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].name == "New Name"
    finally:
        await engine.dispose()


async def test_upsert_staff_coerces_is_on_job():
    """is_on_job '1' -> True; '0' -> False; None preserved."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            await upsert_staff(
                session, {"user_id": "u_b1", "is_on_job": "1"}
            )
            await upsert_staff(
                session, {"user_id": "u_b0", "is_on_job": "0"}
            )
            await upsert_staff(
                session, {"user_id": "u_bn", "is_on_job": None}
            )
            await session.commit()

            on = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "u_b1")
                )
            ).scalar_one()
            off = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "u_b0")
                )
            ).scalar_one()
            none_ = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "u_bn")
                )
            ).scalar_one()
            assert on.is_on_job is True
            assert off.is_on_job is False
            assert none_.is_on_job is None
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# upsert_wechat_account now ALSO upserts the embedded Staff (Task 2.1)
# ---------------------------------------------------------------------------


async def test_upsert_wechat_account_also_upserts_embedded_staff():
    """Account upsert creates BOTH the WechatAccount row and the Staff row."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "wx_id": "wxid_dup",
                "nick_name": "Dup WeChat",
                "user_id": "u_dup",
                "user_name": "Dup Real Name",
                "is_on_job": 1,
                "phone": "13800000099",
            }
            await upsert_wechat_account(session, rec)
            await session.commit()

            acct = (
                await session.execute(
                    sa.select(WechatAccount).where(WechatAccount.wx_id == "wxid_dup")
                )
            ).scalar_one()
            assert acct.user_id == "u_dup"
            assert acct.user_name == "Dup Real Name"

            staff = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "u_dup")
                )
            ).scalar_one()
            assert staff.name == "Dup Real Name"
            assert staff.is_on_job is True
    finally:
        await engine.dispose()


async def test_upsert_wechat_account_reupsert_updates_embedded_staff():
    """Re-upserting the account refreshes BOTH account + embedded staff rows."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "wx_id": "wxid_re",
                "nick_name": "Old Nick",
                "user_id": "u_re",
                "user_name": "Old Name",
                "is_on_job": 1,
            }
            rec2 = {
                "wx_id": "wxid_re",
                "nick_name": "New Nick",
                "user_id": "u_re",
                "user_name": "New Name",
                "is_on_job": 0,  # staff departed since last scrape
            }
            await upsert_wechat_account(session, rec1)
            await session.commit()
            await upsert_wechat_account(session, rec2)
            await session.commit()

            # One account row, refreshed.
            acct_rows = (
                await session.execute(
                    sa.select(WechatAccount)
                    .where(WechatAccount.wx_id == "wxid_re")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(acct_rows) == 1
            assert acct_rows[0].nick_name == "New Nick"

            # One staff row, refreshed (not duplicated).
            staff_rows = (
                await session.execute(
                    sa.select(Staff)
                    .where(Staff.user_id == "u_re")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(staff_rows) == 1
            assert staff_rows[0].name == "New Name"
            assert staff_rows[0].is_on_job is False
    finally:
        await engine.dispose()


async def test_upsert_wechat_account_without_user_id_skips_staff():
    """Account with no user_id is written; embedded-staff upsert is skipped."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "wx_id": "wxid_no_staff",
                "nick_name": "Orphan WeChat",
                # no user_id / user_name
            }
            await upsert_wechat_account(session, rec)
            await session.commit()

            # Account row written.
            acct = (
                await session.execute(
                    sa.select(WechatAccount).where(
                        WechatAccount.wx_id == "wxid_no_staff"
                    )
                )
            ).scalar_one()
            assert acct.nick_name == "Orphan WeChat"

            # No staff rows at all.
            staff_rows = (
                await session.execute(sa.select(Staff))
            ).scalars().all()
            assert staff_rows == []
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# ChatGroup + GroupMember upsert (M2)
# ---------------------------------------------------------------------------


async def test_upsert_chat_group_inserts_group_and_members():
    """Upsert writes the ChatGroup row AND one GroupMember per member_ids_list."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "room_id": "12345678901@chatroom",
                "channel": "wechat",
                "room_name": "VIP客户交流群",
                "owner_account_id": "wxid_staff_alice01",
                "member_count": 3,
                "member_ids": "wxid_m1;wxid_m2;wxid_m3",
                "member_ids_list": ["wxid_m1", "wxid_m2", "wxid_m3"],
                "message_count": 4521,
            }
            await upsert_chat_group(session, rec)
            await session.commit()

            group = (
                await session.execute(
                    sa.select(ChatGroup).where(
                        ChatGroup.room_id == "12345678901@chatroom"
                    )
                )
            ).scalar_one()
            assert group.room_name == "VIP客户交流群"
            assert group.owner_account_id == "wxid_staff_alice01"
            assert group.member_count == 3
            assert group.message_count == 4521
            # The raw semicolon-joined string is stored on the group row.
            assert group.member_ids == "wxid_m1;wxid_m2;wxid_m3"

            # Members: one row per member_ids_list entry, composite PK.
            members = (
                await session.execute(
                    sa.select(GroupMember)
                    .where(GroupMember.room_id == "12345678901@chatroom")
                    .order_by(GroupMember.member_id)
                )
            ).scalars().all()
            assert len(members) == 3
            assert {m.member_id for m in members} == {"wxid_m1", "wxid_m2", "wxid_m3"}
            assert {m.channel for m in members} == {"wechat"}
            # Phase-0: per-member nickname/avatar are not yet populated.
            assert all(m.nickname is None for m in members)
            assert all(m.avatar is None for m in members)
    finally:
        await engine.dispose()


async def test_upsert_chat_group_reupsert_updates_group_keeps_members_unique():
    """Re-upserting a group refreshes its row; members stay unique (no dups)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "room_id": "98765432109@chatroom",
                "channel": "wechat",
                "owner_account_id": "wxid_owner_re",
                "room_name": "Old Name",
                "member_count": 2,
                "member_ids": "wxid_m1;wxid_m2",
                "member_ids_list": ["wxid_m1", "wxid_m2"],
                "message_count": 10,
            }
            await upsert_chat_group(session, rec1)
            await session.commit()

            rec2 = dict(
                rec1,
                room_name="New Name",
                message_count=99,
            )
            await upsert_chat_group(session, rec2)
            await session.commit()

            # One group row, refreshed (name + count updated, not duplicated).
            groups = (
                await session.execute(
                    sa.select(ChatGroup)
                    .where(ChatGroup.room_id == "98765432109@chatroom")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(groups) == 1
            assert groups[0].room_name == "New Name"
            assert groups[0].message_count == 99

            # Members still exactly 2 (idempotent re-upsert, no duplication).
            members = (
                await session.execute(
                    sa.select(GroupMember).where(
                        GroupMember.room_id == "98765432109@chatroom"
                    )
                )
            ).scalars().all()
            assert len(members) == 2
            assert {m.member_id for m in members} == {"wxid_m1", "wxid_m2"}
    finally:
        await engine.dispose()


async def test_upsert_chat_group_drops_empty_member_id():
    """An empty member id in member_ids_list is dropped (no empty-PK row).

    The scraper already drops empties when building ``member_ids_list``; this
    test exercises the repo's own defensive guard so a hand-built record that
    slips an empty id through cannot create a degenerate empty-PK member row.
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "room_id": "55555555555@chatroom",
                "channel": "wechat",
                "owner_account_id": "wxid_owner_empty",
                # Empty string sneaks through; repo must skip it.
                "member_ids_list": ["", "wxid_only", ""],
            }
            await upsert_chat_group(session, rec)
            await session.commit()

            members = (
                await session.execute(
                    sa.select(GroupMember).where(
                        GroupMember.room_id == "55555555555@chatroom"
                    )
                )
            ).scalars().all()
            assert len(members) == 1
            assert members[0].member_id == "wxid_only"
    finally:
        await engine.dispose()


async def test_upsert_chat_group_with_no_members_writes_group_only():
    """A group with an empty member list still writes the group row."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "room_id": "empty@chatroom",
                "channel": "wechat",
                "owner_account_id": "wxid_owner_none",
                "room_name": "No Members Yet",
                "member_ids_list": [],
            }
            await upsert_chat_group(session, rec)
            await session.commit()

            # Group row written.
            group = (
                await session.execute(
                    sa.select(ChatGroup).where(ChatGroup.room_id == "empty@chatroom")
                )
            ).scalar_one()
            assert group.room_name == "No Members Yet"

            # No member rows.
            members = (
                await session.execute(
                    sa.select(GroupMember).where(GroupMember.room_id == "empty@chatroom")
                )
            ).scalars().all()
            assert members == []
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# PhoneCall (voice_call) upsert (Task 2.4)
# ---------------------------------------------------------------------------


async def test_upsert_voice_call_is_idempotent_and_updates_record_file():
    """Same call_id twice with different record_file/version -> 1 row, refreshed."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "call_id": "vc_dup",
                "channel": "voice_call",
                "record_file": "https://example.com/call_v1.mp3",
                "call_duration": 10,
                "version": "v1",
            }
            rec2 = {
                "call_id": "vc_dup",
                "channel": "voice_call",
                "record_file": "https://example.com/call_v2.mp3",
                "call_duration": 12,
                "version": "v2",
            }

            await upsert_voice_call(session, rec1)
            await session.commit()
            await upsert_voice_call(session, rec2)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(PhoneCall)
                    .where(PhoneCall.call_id == "vc_dup")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].call_id == "vc_dup"
            assert rows[0].channel == "voice_call"
            assert rows[0].record_file == "https://example.com/call_v2.mp3"
            assert rows[0].call_duration == 12
            assert rows[0].version == "v2"
    finally:
        await engine.dispose()


async def test_upsert_voice_call_null_record_file_and_coerces_ints():
    """null record_file -> None; numeric-str call_duration/start_call_time/
    call_status/incoming_call coerced to int."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "call_id": "vc_null",
                "channel": "voice_call",
                "record_file": None,
                "call_duration": "151",  # str -> int
                "start_call_time": "1700000000123",  # numeric str -> int (ms epoch)
                "call_status": "1",  # str -> int
                "incoming_call": "1",  # str -> int
            }
            await upsert_voice_call(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(PhoneCall).where(PhoneCall.call_id == "vc_null")
                )
            ).scalar_one()
            assert fetched.record_file is None
            assert fetched.call_duration == 151
            assert fetched.start_call_time == 1700000000123
            assert fetched.call_status == 1
            assert fetched.incoming_call == 1
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# WhatsappAccount + embedded Staff upsert (M3)
# ---------------------------------------------------------------------------


async def test_upsert_whatsapp_account_writes_account_and_embedded_staff():
    """WhatsappAccount upsert creates the account row AND the embedded Staff row."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "uid": "8613800000001@s.whatsapp.net",
                "user_id": "wa_u1",
                "user_name": "Alice Wang",
                "nick_name": "Alice WA",
                "phone": "8613800000001",
                "sale_phone": "8613800000001",
                "sales_bind_phone": "8613900000001",
                "contacts_num": 421,
                "group_num": 12,
                "talker_count": 38,
                "friend_add_num": 7,
                "is_on_job": 1,
                "last_update": 1700000000123,
            }
            await upsert_whatsapp_account(session, rec)
            await session.commit()

            acct = (
                await session.execute(
                    sa.select(WhatsappAccount).where(
                        WhatsappAccount.uid == "8613800000001@s.whatsapp.net"
                    )
                )
            ).scalar_one()
            assert acct.user_id == "wa_u1"
            assert acct.user_name == "Alice Wang"
            assert acct.sale_phone == "8613800000001"
            assert acct.sales_bind_phone == "8613900000001"
            assert acct.contacts_num == 421
            assert acct.is_on_job is True
            assert acct.last_update == 1700000000123

            # Embedded staff row created (PK = user_id), name forwarded.
            staff = (
                await session.execute(
                    sa.select(Staff).where(Staff.user_id == "wa_u1")
                )
            ).scalar_one()
            assert staff.name == "Alice Wang"
            assert staff.is_on_job is True
            assert staff.sale_phone == "8613800000001"
    finally:
        await engine.dispose()


async def test_upsert_whatsapp_account_is_idempotent_and_refreshes_staff():
    """Re-upserting the account refreshes BOTH account + embedded staff rows."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "uid": "8613800000002@s.whatsapp.net",
                "user_id": "wa_u2",
                "user_name": "Old Name",
                "nick_name": "Old Nick",
                "is_on_job": 1,
            }
            rec2 = {
                "uid": "8613800000002@s.whatsapp.net",
                "user_id": "wa_u2",
                "user_name": "New Name",
                "nick_name": "New Nick",
                "is_on_job": 0,  # staff departed since last scrape
            }
            await upsert_whatsapp_account(session, rec1)
            await session.commit()
            await upsert_whatsapp_account(session, rec2)
            await session.commit()

            # One account row, refreshed.
            acct_rows = (
                await session.execute(
                    sa.select(WhatsappAccount)
                    .where(WhatsappAccount.uid == "8613800000002@s.whatsapp.net")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(acct_rows) == 1
            assert acct_rows[0].nick_name == "New Nick"

            # One staff row, refreshed (not duplicated).
            staff_rows = (
                await session.execute(
                    sa.select(Staff)
                    .where(Staff.user_id == "wa_u2")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(staff_rows) == 1
            assert staff_rows[0].name == "New Name"
            assert staff_rows[0].is_on_job is False
    finally:
        await engine.dispose()


async def test_upsert_whatsapp_account_without_user_id_skips_staff():
    """Account with no user_id is written; embedded-staff upsert is skipped."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "uid": "8613800000009@s.whatsapp.net",
                "nick_name": "Orphan WA",
                # no user_id / user_name
            }
            await upsert_whatsapp_account(session, rec)
            await session.commit()

            # Account row written.
            acct = (
                await session.execute(
                    sa.select(WhatsappAccount).where(
                        WhatsappAccount.uid == "8613800000009@s.whatsapp.net"
                    )
                )
            ).scalar_one()
            assert acct.nick_name == "Orphan WA"

            # No staff rows at all.
            staff_rows = (
                await session.execute(sa.select(Staff))
            ).scalars().all()
            assert staff_rows == []
    finally:
        await engine.dispose()


async def test_upsert_customer_with_whatsapp_channel():
    """WA contacts flow through the channel-generic upsert_customer.

    The ``Customer`` table is shared across WeChat / WhatsApp / SMS peers; the
    per-row ``channel`` value disambiguates them. This pins that a WA-shaped
    record (``channel="whatsapp"``) upserts cleanly against the same PK contract.
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "friend_id": "8613800000111@s.whatsapp.net",
                "channel": "whatsapp",
                "owner_account_id": "8613800000001@s.whatsapp.net",
                "friend_nickname": "Customer One",
                "phone": "8613800000111",
                "sale_name": "Alice Sale",
                "user_id": "wa_staff_001",
                "message_count": 57,
            }
            await upsert_customer(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Customer).where(
                        Customer.friend_id == "8613800000111@s.whatsapp.net"
                    )
                )
            ).scalar_one()
            assert fetched.channel == "whatsapp"
            assert fetched.owner_account_id == "8613800000001@s.whatsapp.net"
            assert fetched.friend_nickname == "Customer One"
            assert fetched.sale_name == "Alice Sale"
            assert fetched.message_count == 57
    finally:
        await engine.dispose()


async def test_upsert_messages_with_whatsapp_channel():
    """WA messages flow through the channel-generic upsert_messages."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "msg_id": "wa_msg_1",
                "channel": "whatsapp",
                "owner_account_id": "8613800000001@s.whatsapp.net",
                "talker_id": "8613800000111@s.whatsapp.net",
                "is_send": 0,
                "type": "text",
                "type_raw": 0,
                "text": "hi",
            }
            await upsert_messages(session, [rec])
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Message).where(Message.msg_id == "wa_msg_1")
                )
            ).scalar_one()
            assert fetched.channel == "whatsapp"
            assert fetched.type == "text"
            assert fetched.type_raw == 0
            assert fetched.is_send is False
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# PhoneCall (phone_call) upsert (M4)
# ---------------------------------------------------------------------------


async def test_upsert_phone_call_is_idempotent_and_updates_record_file():
    """Same call_id twice with different record_file/version -> 1 row, refreshed.

    The phone-call channel (``channel="phone_call"``) shares the
    :class:`PhoneCall` table with the voice-call channel; the per-row
    ``channel`` value disambiguates them.
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "call_id": "pc_dup",
                "channel": "phone_call",
                "record_file": "https://example.com/call_v1.mp3",
                "call_duration": 10,
                "call_status": 2,
                "incoming_call": 0,
                "version": "v1",
            }
            rec2 = {
                "call_id": "pc_dup",
                "channel": "phone_call",
                "record_file": "https://example.com/call_v2.mp3",
                "call_duration": 12,
                "call_status": 0,
                "incoming_call": 1,
                "version": "v2",
            }

            await upsert_phone_call(session, rec1)
            await session.commit()
            await upsert_phone_call(session, rec2)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(PhoneCall)
                    .where(PhoneCall.call_id == "pc_dup")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].call_id == "pc_dup"
            assert rows[0].channel == "phone_call"
            assert rows[0].record_file == "https://example.com/call_v2.mp3"
            assert rows[0].call_duration == 12
            assert rows[0].call_status == 0
            assert rows[0].incoming_call == 1
            assert rows[0].version == "v2"
    finally:
        await engine.dispose()


async def test_upsert_phone_call_null_record_file_and_coerces_ints():
    """null record_file -> None; numeric-str call_duration/start_call_time/
    call_status/incoming_call/qc_status coerced to int; is_upload -> bool."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "call_id": "pc_null",
                "channel": "phone_call",
                "record_file": None,
                "call_duration": "151",  # str -> int
                "start_call_time": "1700000000123",  # numeric str -> int (ms epoch)
                "call_status": "2",  # str -> int
                "incoming_call": "1",  # str -> int (1=incoming/0=outgoing)
                "qc_status": "1",  # str -> int
                "is_upload": "1",  # str -> bool
                "download_count": 3,
            }
            await upsert_phone_call(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(PhoneCall).where(PhoneCall.call_id == "pc_null")
                )
            ).scalar_one()
            assert fetched.record_file is None
            assert fetched.call_duration == 151
            assert fetched.start_call_time == 1700000000123
            assert fetched.call_status == 2
            assert fetched.incoming_call == 1
            assert fetched.qc_status == 1
            assert fetched.is_upload is True
            assert fetched.download_count == 3
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# Sms upsert (M4)
# ---------------------------------------------------------------------------


async def test_upsert_sms_is_idempotent_and_updates_content():
    """Same sms_id twice with different content/version -> 1 row, refreshed."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "sms_id": "sms_dup",
                "channel": "sms",
                "customer_phone": "8613800000001",
                "content": "first message",
                "content_preview": "first...",
                "direction": 0,
                "create_time": 1_700_000_000_000,
                "version": "v1",
            }
            rec2 = {
                "sms_id": "sms_dup",
                "channel": "sms",
                "customer_phone": "8613800000001",
                "content": "updated message",
                "content_preview": "updated...",
                "direction": 0,
                "create_time": 1_700_000_000_000,
                "version": "v2",
            }

            await upsert_sms(session, rec1)
            await session.commit()
            await upsert_sms(session, rec2)
            await session.commit()

            rows = (
                await session.execute(
                    sa.select(Sms)
                    .where(Sms.sms_id == "sms_dup")
                    .execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].sms_id == "sms_dup"
            assert rows[0].channel == "sms"
            assert rows[0].content == "updated message"
            assert rows[0].content_preview == "updated..."
            assert rows[0].version == "v2"
    finally:
        await engine.dispose()


async def test_upsert_sms_coerces_direction_and_create_time():
    """direction numeric str -> int; create_time numeric str -> int (ms epoch)."""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec = {
                "sms_id": "sms_coerce",
                "channel": "sms",
                "direction": "1",  # str -> int (1=发/outgoing)
                "create_time": "1700000000123",  # numeric str -> int (ms epoch)
                "content": "hi",
            }
            await upsert_sms(session, rec)
            await session.commit()

            fetched = (
                await session.execute(
                    sa.select(Sms).where(Sms.sms_id == "sms_coerce")
                )
            ).scalar_one()
            assert fetched.direction == 1
            assert fetched.create_time == 1700000000123
            assert fetched.content == "hi"
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# MomentsPost upsert (员工聊天窗口 Phase 1)
# ---------------------------------------------------------------------------


async def test_upsert_moments_posts_idempotent_keeps_json_lists_and_coerces_create_time():
    """MomentsPost batch upsert: idempotent on PK ``id``; JSON list columns
    (praise_json / comment_json / image_urls) stay real lists (NOT joined into
    ";..." strings by the scalar-column list coercion); numeric-str
    create_time coerces to ms-epoch int.
    """
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            rec1 = {
                "id": "69e082a3c3524a7fe3ac53b4",
                "wechat_id": "wxid_x",
                "sns_id": "-3546042415332912522",
                "author_name": "Yoyo",
                "author_wxid": "wxid_x",
                "content": "v1 content",
                "type_raw": "1",
                "create_time": 1776302051000,  # ms epoch int
                "praise_json": [{"name": "E**U"}],
                "comment_json": [],
                "image_urls": ["https://wx.oss.example/a.jpg"],
            }
            # Same PK again: later write wins, create_time arrives as a
            # numeric string (defensive coercion path).
            rec2 = dict(rec1, content="v2 content", create_time="1776302051999")
            count = await upsert_moments_posts(session, [rec1, rec2])
            await session.commit()

            assert count == 2  # returns input length
            rows = (
                await session.execute(
                    sa.select(MomentsPost).execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1  # same PK twice -> ONE row
            assert rows[0].content == "v2 content"  # later record wins
            assert rows[0].create_time == 1776302051999  # str -> int
            # JSON list columns preserved verbatim as lists.
            assert rows[0].praise_json == [{"name": "E**U"}]
            assert rows[0].comment_json == []
            assert rows[0].image_urls == ["https://wx.oss.example/a.jpg"]
            assert rows[0].type_raw == "1"  # MomentsPost type_raw is a string code
    finally:
        await engine.dispose()


# ---------------------------------------------------------------------------
# int 强转 vs String 列声明(2026-08-25 评审遗留修复)
# _INT_FIELDS 按列名全局命中,但列类型优先:目标列声明为 String(Text 继承)
# 则跳过强转 —— MomentsPost.type_raw 是 String(16),非数字 code(如 "text")
# 不能被 _coerce_int 置 None;Message.type_raw(Integer)照旧强转。
# ---------------------------------------------------------------------------


async def test_int_coercion_skipped_for_string_columns():
    """String 声明的 type_raw 原样保留;Integer 声明的照旧强转。"""
    # MomentsPost.type_raw = String(16) -> 不强转
    vals = _build_column_values(
        MomentsPost, {"id": "x", "wechat_id": "w", "type_raw": "1"}
    )
    assert vals["type_raw"] == "1"
    vals = _build_column_values(
        MomentsPost, {"id": "x", "wechat_id": "w", "type_raw": "text"}
    )
    assert vals["type_raw"] == "text"

    # Message.type_raw = Integer -> "1" -> 1,非数字 -> None
    vals = _build_column_values(
        Message, {"msg_id": "m", "channel": "wechat", "type_raw": "1"}
    )
    assert vals["type_raw"] == 1
    vals = _build_column_values(
        Message, {"msg_id": "m", "channel": "wechat", "type_raw": "text"}
    )
    assert vals["type_raw"] is None


async def test_moments_post_non_numeric_type_raw_round_trip():
    """非数字 type_raw code 入库读回仍是原字符串(不被 _coerce_int 置 None)。"""
    engine = await _make_engine()
    try:
        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
        async with SessionLocal() as session:
            await upsert_moments_posts(
                session,
                [
                    {
                        "id": "p1",
                        "wechat_id": "wxid_x",
                        "type_raw": "video",  # 假设未来出现的非数字 code
                        "create_time": 1776302051000,
                    }
                ],
            )
            await session.commit()
            rows = (
                await session.execute(
                    sa.select(MomentsPost).execution_options(populate_existing=True)
                )
            ).scalars().all()
            assert len(rows) == 1
            assert rows[0].type_raw == "video"
    finally:
        await engine.dispose()