first commit

tests/test_auth_store.py

@@ -0,0 +1,166 @@
+"""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()