"""chongming-auth 共享包测试：schema 兼容、seed 幂等/并发、登录校验、claims 契约。"""
import asyncio
import tempfile
from pathlib import Path

import pytest
from chongming_jwt import JWT

from chongming_auth import (
    LoginPostRequest,
    LoginPostResponse,
    Users,
    issue_token,
    open_database,
    seed_admin,
    touch_last_login,
    verify_login,
)


@pytest.fixture()
def anyio_backend():
    return "asyncio"


@pytest.fixture()
async def db(tmp_path):
    database = await open_database({"data_dir": str(tmp_path)})
    yield database
    await database.dispose()


@pytest.mark.anyio
async def test_open_database_creates_users_table(tmp_path):
    """建库即建 users 表（SQLModel 元数据随包导入注册）。"""
    database = await open_database({"data_dir": str(tmp_path)})
    async with database.get_session() as session:
        from sqlalchemy import select

        rows = (await session.execute(select(Users))).scalars().all()
    assert rows == []
    await database.dispose()


@pytest.mark.anyio
async def test_seed_admin_idempotent(db):
    assert await seed_admin(db) is True
    # 重复种子（按 username / 按 email 兜底）均不再新建
    assert await seed_admin(db) is False
    assert await seed_admin(db, password="other") is False


@pytest.mark.anyio
async def test_seed_admin_concurrent(tmp_path):
    """gunicorn 多进程并发启动场景：并发种子不抛 IntegrityError，仅一条 admin。"""
    database = await open_database({"data_dir": str(tmp_path)})
    results = await asyncio.gather(
        *[seed_admin(database) for _ in range(8)], return_exceptions=True
    )
    errors = [r for r in results if isinstance(r, Exception)]
    assert not errors, f"并发种子存在未捕获异常: {errors}"
    assert sum(1 for r in results if r is True) == 1

    from sqlalchemy import select

    async with database.get_session() as session:
        rows = (await session.execute(select(Users))).scalars().all()
    assert len(rows) == 1
    assert rows[0].is_superuser is True
    await database.dispose()


@pytest.mark.anyio
async def test_verify_login_messages(db):
    """失败文案与拆分前 plugins.auth.login 逐字一致（前端兼容）。"""
    await seed_admin(db)

    user, msg = await verify_login(db, "admin", "admin")
    assert user is not None and msg == ""

    user, msg = await verify_login(db, "nouser", "x")
    assert user is None and msg == "用户名错误"

    user, msg = await verify_login(db, "admin", "wrong")
    assert user is None and msg == "密码错误"

    # 软禁用用户
    from chongming_password import hash_password

    async with db.get_session() as session:
        disabled = Users(
            username="zhang",
            email="zhang@example.com",
            hashed_password=hash_password("pwd"),
            is_active=False,
        )
        session.add(disabled)
        await session.commit()
    user, msg = await verify_login(db, "zhang", "pwd")
    assert user is None and msg == "用户已经注销或被禁用"


@pytest.mark.anyio
async def test_issue_token_claims_contract(db):
    """claims 与拆分前 plugins.auth.login 完全一致，且可被网关 JWT 解密回读。"""
    await seed_admin(db)
    user, _ = await verify_login(db, "admin", "admin")

    secret, algorithm = "test-secret", "HS256"
    token = issue_token(user, secret, algorithm, expires_in_seconds=60)

    payload = __import__("orjson").loads(JWT(secret, algorithm).decode(token))
    # 业务 claims 全集 + chongming_jwt 注入的标准注册声明（iat/exp，拆分前亦然）
    assert {"sub", "username", "email", "full_name", "avatar_url", "phone",
            "is_verified", "is_superuser", "created_at", "updated_at",
            "last_login_at"} <= set(payload.keys())
    assert {"iat", "exp"} <= set(payload.keys())
    assert payload["sub"] == user.id
    assert payload["username"] == "admin"
    assert payload["is_superuser"] is True


@pytest.mark.anyio
async def test_touch_last_login_persists(db):
    await seed_admin(db)
    user, _ = await verify_login(db, "admin", "admin")
    assert user.last_login_at is None

    await touch_last_login(db, user)

    from sqlalchemy import select

    async with db.get_session() as session:
        row = (await session.execute(select(Users).filter_by(username="admin"))).scalar_one()
    assert row.last_login_at is not None


def test_login_contract_models_shape():
    """请求/响应模型字段与拆分前 gen_models 产物一致。"""
    assert set(LoginPostRequest.model_fields.keys()) == {"username", "password"}
    assert set(LoginPostResponse.model_fields.keys()) == {
        "result", "token", "message", "timestamp"
    }
    resp = LoginPostResponse(result="fail", token="", message="用户名错误")
    assert resp.timestamp is None


def test_users_schema_matches_legacy(tmp_path):
    """对既有 auth.db 的兼容：老库打开后 users 表结构不变、数据可读。"""
    # 用 tempfile 模拟一个已有库（先建 → 关 → 重开）
    data_dir = tmp_path / "data"
    data_dir.mkdir()
    asyncio.run(_build_then_reopen(data_dir))


async def _build_then_reopen(data_dir: Path):
    db1 = await open_database({"data_dir": str(data_dir)})
    await seed_admin(db1)
    await db1.dispose()

    db2 = await open_database({"data_dir": str(data_dir)})
    created = await seed_admin(db2)
    assert created is False
    user, msg = await verify_login(db2, "admin", "admin")
    assert user is not None and msg == ""
    await db2.dispose()