chongming-core/test/test_auth_oauth.py

"""
auth_oauth 身份源插件测试（基于 chongming-auth 共享契约的第三方登录示例）

覆盖:
  1. OTP 手机验证码登录全流程（演示模式 dev_code）：
     发码 → 错码拒绝 → 正确码登录签发平台 token → 同号复登复用同一账号（绑定生效）
  2. GitHub OAuth 授权码流程（HTTP 调用打桩）：
     authorize 生成 URL+state → callback find-or-create 签发 token → state 一次性（重放拒绝）
  3. 绑定表与共享 users 表落库验证（chongming_auth.Users 可查）
  4. 插件导入 + 三方契约校验 0 错误
"""
import asyncio
import tempfile
from pathlib import Path

import orjson
import pytest
from chongming_auth import Users, open_database
from chongming_cache import Cache
from chongming_config import get_config
from chongming_jwt import JWT
from chongming_worker.contracts import validate_worker_contracts

# 与被测插件同源：issue_token 从 get_config()["security"] 读密钥，测试解码必须一致
_SECURITY = get_config()["security"]
SECRET, ALGORITHM = _SECURITY["secret_key"], _SECURITY["algorithm"]


@pytest.fixture()
def env(tmp_path: Path):
    """初始化插件运行环境：共享 auth 库 + 临时 sqlite 缓存 + global_var 注入。

    注意：插件模块 @rpc_method 注册在进程内只执行一次，fixture 不清空注册表
    （clear_registry 会导致后续测试 import no-op、注册表为空）。
    """
    import plugins.auth_oauth  # noqa: F401  触发 @rpc_method 注册（幂等）
    import plugins.auth_oauth.global_var as gv

    async def _build():
        db = await open_database({"data_dir": str(tmp_path)})
        cache = Cache(backend="sqlite", path=str(tmp_path / "cache.sqlite"))
        return db, cache

    db, cache = asyncio.run(_build())
    gv.set_global_var("db", db)
    gv.set_global_var("cache", cache)
    gv.set_global_var("plugin_config", {
        "return_otp_in_response": True,
        "github": {
            "client_id": "test-client-id",
            "client_secret": "test-client-secret",
            "redirect_uri": "http://localhost:5173/auth/oauth/github/callback",
            "scope": "read:user user:email",
        },
    })

    yield gv

    async def _teardown():
        await db.dispose()
        cache.close()

    asyncio.run(_teardown())


def _decode(token: str) -> dict:
    return orjson.loads(JWT(SECRET, ALGORITHM).decode(token))


# ---------- 手机验证码登录 ----------

def test_otp_full_flow(env):
    from plugins.auth_oauth.router.phone import (
        PhoneLoginPostRequest,
        PhoneSendCodePostRequest,
        phone_login,
        phone_send_code,
    )

    phone = "13800138000"

    # 1. 发码（演示模式返回 dev_code）
    sent = asyncio.run(phone_send_code(PhoneSendCodePostRequest(phone=phone)))
    assert sent.result == "success" and sent.dev_code

    # 2. 错码拒绝
    wrong = asyncio.run(phone_login(
        PhoneLoginPostRequest(phone=phone, code="000000" if sent.dev_code != "000000" else "111111")
    ))
    assert wrong.result == "fail" and wrong.token == ""

    # 3. 正确码登录 → 平台 token
    ok = asyncio.run(phone_login(
        PhoneLoginPostRequest(phone=phone, code=sent.dev_code)
    ))
    assert ok.result == "success" and ok.token
    claims = _decode(ok.token)
    assert claims["username"] == phone
    assert claims["is_superuser"] is False

    # 4. 验证码一次性
    replay = asyncio.run(phone_login(
        PhoneLoginPostRequest(phone=phone, code=sent.dev_code)
    ))
    assert replay.result == "fail"

    # 5. 复登（再次发码登录）→ 复用同一账号（绑定生效，不重复建号）
    sent2 = asyncio.run(phone_send_code(PhoneSendCodePostRequest(phone=phone)))
    ok2 = asyncio.run(phone_login(
        PhoneLoginPostRequest(phone=phone, code=sent2.dev_code)
    ))
    assert _decode(ok2.token)["sub"] == claims["sub"]


def test_otp_invalid_phone(env):
    from plugins.auth_oauth.router.phone import (
        PhoneSendCodePostRequest, phone_send_code,
    )

    resp = asyncio.run(phone_send_code(PhoneSendCodePostRequest(phone="abc")))
    assert resp.result == "fail" and "手机号" in resp.message


# ---------- GitHub OAuth ----------

def test_github_authorize_and_callback(env, monkeypatch):
    from plugins.auth_oauth.router.github import (
        GithubCallbackPostRequest, github_authorize, github_callback,
    )
    from plugins.auth_oauth.providers import github as github_provider

    # authorize：URL 含 client_id 与 state，state 已入缓存
    authz = asyncio.run(github_authorize())
    assert authz.result == "success"
    assert "client_id=test-client-id" in authz.authorize_url
    assert f"state={authz.state}" in authz.authorize_url
    assert env.cache.get(f"auth_oauth:state:{authz.state}")

    # 打桩：授权码换取与用户拉取（免外网）
    fake_profile = {
        "uid": "97594220", "username": "chongming-demo",
        "email": "demo@example.com", "full_name": "Chongming Demo",
        "avatar_url": "https://avatars.githubusercontent.com/u/97594220",
    }
    exchanged = {}

    async def fake_exchange(client_id, client_secret, code, redirect_uri):
        exchanged["code"] = code
        return "fake-access-token"

    async def fake_fetch(access_token):
        assert access_token == "fake-access-token"
        return dict(fake_profile)

    monkeypatch.setattr(github_provider, "exchange_code", fake_exchange)
    monkeypatch.setattr(github_provider, "fetch_user", fake_fetch)

    resp = asyncio.run(github_callback(
        GithubCallbackPostRequest(code="gh-code", state=authz.state)
    ))
    assert resp.result == "success" and resp.token
    claims = _decode(resp.token)
    assert claims["username"] == "chongming-demo"
    assert claims["email"] == "demo@example.com"
    assert exchanged["code"] == "gh-code"

    # state 一次性：重放拒绝
    replay = asyncio.run(github_callback(
        GithubCallbackPostRequest(code="gh-code", state=authz.state)
    ))
    assert replay.result == "fail" and "state" in replay.message

    # 落库验证：users 表有新账号 + 绑定指向同一账号；再次登录（新 state）复用
    async def _read_users():
        db = env.db
        from sqlalchemy import select

        async with db.get_session() as sess:
            return (await sess.execute(select(Users))).scalars().all()

    users = asyncio.run(_read_users())
    assert len(users) == 1
    assert users[0].email == "demo@example.com"

    async def _second_login():
        from plugins.auth_oauth.router.github import github_authorize as _a

        a = await _a()
        return await github_callback(
            GithubCallbackPostRequest(code="gh-code-2", state=a.state)
        )

    second = asyncio.run(_second_login())
    assert _decode(second.token)["sub"] == claims["sub"]


def test_github_callback_bad_state(env):
    from plugins.auth_oauth.router.github import (
        GithubCallbackPostRequest, github_callback,
    )

    resp = asyncio.run(github_callback(
        GithubCallbackPostRequest(code="x", state="not-exist")
    ))
    assert resp.result == "fail" and "state" in resp.message


# ---------- 契约校验 ----------

def test_contract_validation(env):
    import plugins.auth_oauth  # noqa: F401  确保注册

    issues = validate_worker_contracts("auth_oauth")
    errors = [i for i in issues if i.get("level") == "error"]
    assert errors == [], f"契约校验存在错误: {errors}"