""" 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}"