216 lines
7.4 KiB
Python
216 lines
7.4 KiB
Python
"""
|
||
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}"
|