Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| eba067088e | |||
| ab6a2ffb08 | |||
| 3be0bc68a0 | |||
| 1883d9ba06 | |||
| 2b883b5468 | |||
| 5e0c2f2654 | |||
| 550b6cb73c | |||
| 6c5e76d32e | |||
| b1e85b68df | |||
| 21a66be8a6 | |||
| 4a848247fa | |||
| af38a296c5 | |||
| f021ff2a74 | |||
| 0fc133798b | |||
| 61117e50bc | |||
| ad902105f5 | |||
| f83c96bb77 | |||
| 6930d4543f | |||
| 7b2f35ad3d | |||
| 5256bd0d8b | |||
| 33270dd026 |
@@ -1,188 +0,0 @@
|
||||
/**
|
||||
* 认证相关 API
|
||||
*/
|
||||
import axios from "axios"
|
||||
import apiClient from "./client"
|
||||
|
||||
// 类型定义
|
||||
export interface LoginRequest {
|
||||
email: string
|
||||
password: string
|
||||
}
|
||||
|
||||
export interface LoginResponse {
|
||||
access_token: string
|
||||
refresh_token?: string | null
|
||||
token_type: string
|
||||
expires_in: number
|
||||
user_id: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
}
|
||||
|
||||
export interface RegisterRequest {
|
||||
email: string
|
||||
password: string
|
||||
username: string
|
||||
display_name?: string
|
||||
}
|
||||
|
||||
export interface User {
|
||||
id: string
|
||||
user_id: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
is_email_verified: boolean
|
||||
email_verified: boolean
|
||||
created_at?: string
|
||||
}
|
||||
|
||||
export interface UserResponse {
|
||||
id?: string
|
||||
user_id?: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
is_email_verified?: boolean
|
||||
email_verified?: boolean
|
||||
created_at?: string
|
||||
}
|
||||
|
||||
export const normalizeUser = (data: UserResponse): User => {
|
||||
const userId = data.id ?? data.user_id ?? ""
|
||||
const emailVerified = data.is_email_verified ?? data.email_verified ?? false
|
||||
|
||||
return {
|
||||
id: userId,
|
||||
user_id: userId,
|
||||
email: data.email,
|
||||
username: data.username,
|
||||
display_name: data.display_name,
|
||||
is_email_verified: emailVerified,
|
||||
email_verified: emailVerified,
|
||||
created_at: data.created_at,
|
||||
}
|
||||
}
|
||||
|
||||
// 登录
|
||||
export const login = async (data: LoginRequest): Promise<LoginResponse> => {
|
||||
const response = await apiClient.post("/auth/login", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 刷新 access_token(使用裸 axios 避免拦截器递归)
|
||||
export const refreshAccessToken = async (refreshToken: string): Promise<LoginResponse> => {
|
||||
const baseURL = apiClient.defaults.baseURL ?? ""
|
||||
const response = await axios.post(`${baseURL}/auth/refresh`, {
|
||||
refresh_token: refreshToken,
|
||||
})
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 注册
|
||||
export const register = async (data: RegisterRequest): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/register", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 登出
|
||||
export const logout = async (): Promise<void> => {
|
||||
await apiClient.post("/auth/logout")
|
||||
}
|
||||
|
||||
// 获取当前用户
|
||||
export const getCurrentUser = async (): Promise<User> => {
|
||||
const response = await apiClient.get<UserResponse>("/auth/me")
|
||||
return normalizeUser(response.data)
|
||||
}
|
||||
|
||||
// 请求密码重置
|
||||
export const requestPasswordReset = async (email: string): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/forgot-password", { email })
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 重置密码
|
||||
export const resetPassword = async (
|
||||
token: string,
|
||||
newPassword: string,
|
||||
): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/reset-password", {
|
||||
token,
|
||||
new_password: newPassword,
|
||||
})
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 验证邮箱
|
||||
export const verifyEmail = async (token: string): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/verify-email", { token })
|
||||
return response.data
|
||||
}
|
||||
|
||||
/* ========== 微信登录 ========== */
|
||||
|
||||
export interface WechatAuthUrlResponse {
|
||||
auth_url: string
|
||||
state: string
|
||||
}
|
||||
|
||||
export interface WechatCallbackResponse {
|
||||
access_token: string
|
||||
refresh_token?: string | null
|
||||
user_id: string
|
||||
display_name: string
|
||||
avatar_url: string
|
||||
is_new_user: boolean
|
||||
binding_complete: boolean
|
||||
expires_in: number
|
||||
}
|
||||
|
||||
export interface SendVerificationCodeRequest {
|
||||
target: "email" | "phone"
|
||||
value: string
|
||||
purpose: "bind" | "login" | "reset_password"
|
||||
}
|
||||
|
||||
export interface BindContactRequest {
|
||||
email?: string
|
||||
email_code?: string
|
||||
phone?: string
|
||||
phone_code?: string
|
||||
}
|
||||
|
||||
export interface BindContactResponse {
|
||||
success: boolean
|
||||
user: User
|
||||
}
|
||||
|
||||
// 获取微信授权链接
|
||||
export const getWechatAuthUrl = async (): Promise<WechatAuthUrlResponse> => {
|
||||
const response = await apiClient.get("/auth/wechat/url")
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 微信回调登录
|
||||
export const wechatCallback = async (
|
||||
code: string,
|
||||
state: string,
|
||||
): Promise<WechatCallbackResponse> => {
|
||||
const response = await apiClient.post("/auth/wechat/callback", { code, state })
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 发送验证码
|
||||
export const sendVerificationCode = async (
|
||||
data: SendVerificationCodeRequest,
|
||||
): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/send-verification-code", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
// 绑定联系方式
|
||||
export const bindContact = async (data: BindContactRequest): Promise<BindContactResponse> => {
|
||||
const response = await apiClient.post("/auth/bind-contact", data)
|
||||
return response.data
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import apiClient from "../client"
|
||||
import type { SendVerificationCodeRequest, BindContactRequest, BindContactResponse } from "./types"
|
||||
|
||||
/**
|
||||
* 发送验证码
|
||||
*/
|
||||
export const sendVerificationCode = async (
|
||||
data: SendVerificationCodeRequest,
|
||||
): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/send-verification-code", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 绑定联系方式
|
||||
*/
|
||||
export const bindContact = async (data: BindContactRequest): Promise<BindContactResponse> => {
|
||||
const response = await apiClient.post("/auth/bind-contact", data)
|
||||
return response.data
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
import apiClient from "../client"
|
||||
import type { User, UserResponse } from "./types"
|
||||
import { normalizeUser } from "./user"
|
||||
|
||||
/**
|
||||
* 获取当前用户
|
||||
*/
|
||||
export const getCurrentUser = async (): Promise<User> => {
|
||||
const response = await apiClient.get<UserResponse>("/auth/me")
|
||||
return normalizeUser(response.data)
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
import apiClient from "../client"
|
||||
|
||||
/**
|
||||
* 验证邮箱
|
||||
*/
|
||||
export const verifyEmail = async (token: string): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/verify-email", { token })
|
||||
return response.data
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
/**
|
||||
* 认证相关 API
|
||||
* 保持向后兼容,从子模块 re-export
|
||||
*/
|
||||
|
||||
// 类型
|
||||
export type {
|
||||
LoginRequest,
|
||||
LoginResponse,
|
||||
RegisterRequest,
|
||||
User,
|
||||
UserResponse,
|
||||
WechatAuthUrlResponse,
|
||||
WechatCallbackResponse,
|
||||
SendVerificationCodeRequest,
|
||||
BindContactRequest,
|
||||
BindContactResponse,
|
||||
} from "./types"
|
||||
|
||||
// 用户工具函数
|
||||
export { normalizeUser } from "./user"
|
||||
|
||||
// 登录/注册/登出/刷新
|
||||
export { login, refreshAccessToken, register, logout } from "./login"
|
||||
|
||||
// 当前用户
|
||||
export { getCurrentUser } from "./currentUser"
|
||||
|
||||
// 密码重置
|
||||
export { requestPasswordReset, resetPassword } from "./password"
|
||||
|
||||
// 邮箱验证
|
||||
export { verifyEmail } from "./email"
|
||||
|
||||
// 微信登录
|
||||
export { getWechatAuthUrl, wechatCallback } from "./wechat"
|
||||
|
||||
// 联系方式
|
||||
export { sendVerificationCode, bindContact } from "./contact"
|
||||
@@ -0,0 +1,37 @@
|
||||
import axios from "axios"
|
||||
import apiClient from "../client"
|
||||
import type { LoginRequest, LoginResponse, RegisterRequest } from "./types"
|
||||
|
||||
/**
|
||||
* 登录
|
||||
*/
|
||||
export const login = async (data: LoginRequest): Promise<LoginResponse> => {
|
||||
const response = await apiClient.post("/auth/login", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 刷新 access_token(使用裸 axios 避免拦截器递归)
|
||||
*/
|
||||
export const refreshAccessToken = async (refreshToken: string): Promise<LoginResponse> => {
|
||||
const baseURL = apiClient.defaults.baseURL ?? ""
|
||||
const response = await axios.post(`${baseURL}/auth/refresh`, {
|
||||
refresh_token: refreshToken,
|
||||
})
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 注册
|
||||
*/
|
||||
export const register = async (data: RegisterRequest): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/register", data)
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 登出
|
||||
*/
|
||||
export const logout = async (): Promise<void> => {
|
||||
await apiClient.post("/auth/logout")
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import apiClient from "../client"
|
||||
|
||||
/**
|
||||
* 请求密码重置
|
||||
*/
|
||||
export const requestPasswordReset = async (email: string): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/forgot-password", { email })
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 重置密码
|
||||
*/
|
||||
export const resetPassword = async (
|
||||
token: string,
|
||||
newPassword: string,
|
||||
): Promise<{ message: string }> => {
|
||||
const response = await apiClient.post("/auth/reset-password", {
|
||||
token,
|
||||
new_password: newPassword,
|
||||
})
|
||||
return response.data
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
/**
|
||||
* 认证相关类型定义
|
||||
*/
|
||||
|
||||
export interface LoginRequest {
|
||||
email: string
|
||||
password: string
|
||||
}
|
||||
|
||||
export interface LoginResponse {
|
||||
access_token: string
|
||||
refresh_token?: string | null
|
||||
token_type: string
|
||||
expires_in: number
|
||||
user_id: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
}
|
||||
|
||||
export interface RegisterRequest {
|
||||
email: string
|
||||
password: string
|
||||
username: string
|
||||
display_name?: string
|
||||
}
|
||||
|
||||
export interface User {
|
||||
id: string
|
||||
user_id: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
is_email_verified: boolean
|
||||
email_verified: boolean
|
||||
created_at?: string
|
||||
}
|
||||
|
||||
export interface UserResponse {
|
||||
id?: string
|
||||
user_id?: string
|
||||
email: string
|
||||
username: string
|
||||
display_name: string
|
||||
is_email_verified?: boolean
|
||||
email_verified?: boolean
|
||||
created_at?: string
|
||||
}
|
||||
|
||||
export interface WechatAuthUrlResponse {
|
||||
auth_url: string
|
||||
state: string
|
||||
}
|
||||
|
||||
export interface WechatCallbackResponse {
|
||||
access_token: string
|
||||
refresh_token?: string | null
|
||||
user_id: string
|
||||
display_name: string
|
||||
avatar_url: string
|
||||
is_new_user: boolean
|
||||
binding_complete: boolean
|
||||
expires_in: number
|
||||
}
|
||||
|
||||
export interface SendVerificationCodeRequest {
|
||||
target: "email" | "phone"
|
||||
value: string
|
||||
purpose: "bind" | "login" | "reset_password"
|
||||
}
|
||||
|
||||
export interface BindContactRequest {
|
||||
email?: string
|
||||
email_code?: string
|
||||
phone?: string
|
||||
phone_code?: string
|
||||
}
|
||||
|
||||
export interface BindContactResponse {
|
||||
success: boolean
|
||||
user: User
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import type { User, UserResponse } from "./types"
|
||||
|
||||
/**
|
||||
* 规范化用户数据,兼容不同后端返回格式
|
||||
*/
|
||||
export const normalizeUser = (data: UserResponse): User => {
|
||||
const userId = data.id ?? data.user_id ?? ""
|
||||
const emailVerified = data.is_email_verified ?? data.email_verified ?? false
|
||||
|
||||
return {
|
||||
id: userId,
|
||||
user_id: userId,
|
||||
email: data.email,
|
||||
username: data.username,
|
||||
display_name: data.display_name,
|
||||
is_email_verified: emailVerified,
|
||||
email_verified: emailVerified,
|
||||
created_at: data.created_at,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import apiClient from "../client"
|
||||
import type { WechatAuthUrlResponse, WechatCallbackResponse } from "./types"
|
||||
|
||||
/**
|
||||
* 获取微信授权链接
|
||||
*/
|
||||
export const getWechatAuthUrl = async (): Promise<WechatAuthUrlResponse> => {
|
||||
const response = await apiClient.get("/auth/wechat/url")
|
||||
return response.data
|
||||
}
|
||||
|
||||
/**
|
||||
* 微信回调登录
|
||||
*/
|
||||
export const wechatCallback = async (
|
||||
code: string,
|
||||
state: string,
|
||||
): Promise<WechatCallbackResponse> => {
|
||||
const response = await apiClient.post("/auth/wechat/callback", { code, state })
|
||||
return response.data
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
import { useState, useEffect } from "react"
|
||||
import { Form, message } from "antd"
|
||||
import { sendVerificationCode, bindContact, type BindContactResponse } from "@/api/auth"
|
||||
import { useCountdown } from "./useCountdown"
|
||||
|
||||
type ContactType = "email" | "phone"
|
||||
|
||||
interface UseBindContactFormOptions {
|
||||
open: boolean
|
||||
onSuccess?: (user: BindContactResponse["user"]) => void
|
||||
onCancel?: () => void
|
||||
}
|
||||
|
||||
/**
|
||||
* 绑定联系方式表单 Hook
|
||||
* 封装表单状态、验证码发送、提交绑定等业务逻辑
|
||||
*/
|
||||
export const useBindContactForm = ({ open, onSuccess }: UseBindContactFormOptions) => {
|
||||
const [activeTab, setActiveTab] = useState<ContactType>("email")
|
||||
const [form] = Form.useForm()
|
||||
const [loading, setLoading] = useState(false)
|
||||
const [codeLoading, setCodeLoading] = useState(false)
|
||||
const { countdown, isRunning, start: startCountdown, reset: resetCountdown } = useCountdown(60)
|
||||
|
||||
/* ── 打开时重置表单 ── */
|
||||
useEffect(() => {
|
||||
if (open) {
|
||||
form.resetFields()
|
||||
resetCountdown()
|
||||
}
|
||||
}, [open, form, resetCountdown])
|
||||
|
||||
/* ── 发送验证码 ── */
|
||||
const handleSendCode = async () => {
|
||||
try {
|
||||
const fieldName = activeTab === "email" ? "email" : "phone"
|
||||
const value = form.getFieldValue(fieldName)
|
||||
if (!value) {
|
||||
message.warning(activeTab === "email" ? "请输入邮箱" : "请输入手机号")
|
||||
return
|
||||
}
|
||||
setCodeLoading(true)
|
||||
await sendVerificationCode({
|
||||
target: activeTab,
|
||||
value,
|
||||
purpose: "bind",
|
||||
})
|
||||
message.success("验证码已发送")
|
||||
startCountdown()
|
||||
} catch {
|
||||
// error handled by interceptor
|
||||
} finally {
|
||||
setCodeLoading(false)
|
||||
}
|
||||
}
|
||||
|
||||
/* ── 提交绑定 ── */
|
||||
const handleSubmit = async () => {
|
||||
// 1. 表单验证(失败时 Ant Design Form 会自动展示错误信息)
|
||||
let values
|
||||
try {
|
||||
values = await form.validateFields()
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
|
||||
// 2. 调用绑定接口
|
||||
setLoading(true)
|
||||
try {
|
||||
const payload =
|
||||
activeTab === "email"
|
||||
? { email: values.email, email_code: values.email_code }
|
||||
: { phone: values.phone, phone_code: values.phone_code }
|
||||
|
||||
const result = await bindContact(payload)
|
||||
|
||||
message.success("绑定成功")
|
||||
onSuccess?.(result.user)
|
||||
} catch {
|
||||
// API 错误由全局拦截器统一处理提示
|
||||
} finally {
|
||||
setLoading(false)
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
activeTab,
|
||||
setActiveTab,
|
||||
form,
|
||||
loading,
|
||||
codeLoading,
|
||||
countdown,
|
||||
codeDisabled: isRunning,
|
||||
handleSendCode,
|
||||
handleSubmit,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
import { useState, useEffect, useRef, useCallback } from "react"
|
||||
|
||||
/**
|
||||
* 倒计时 Hook
|
||||
* 用于验证码发送等场景的倒计时控制
|
||||
*/
|
||||
export const useCountdown = (initialSeconds = 60) => {
|
||||
const [countdown, setCountdown] = useState(0)
|
||||
const timerRef = useRef<ReturnType<typeof setInterval> | null>(null)
|
||||
|
||||
useEffect(() => {
|
||||
if (countdown > 0) {
|
||||
timerRef.current = setInterval(() => {
|
||||
setCountdown((prev) => prev - 1)
|
||||
}, 1000)
|
||||
} else if (timerRef.current) {
|
||||
clearInterval(timerRef.current)
|
||||
timerRef.current = null
|
||||
}
|
||||
return () => {
|
||||
if (timerRef.current) clearInterval(timerRef.current)
|
||||
}
|
||||
}, [countdown])
|
||||
|
||||
const start = useCallback(() => {
|
||||
setCountdown(initialSeconds)
|
||||
}, [initialSeconds])
|
||||
|
||||
const reset = useCallback(() => {
|
||||
setCountdown(0)
|
||||
}, [])
|
||||
|
||||
return {
|
||||
countdown,
|
||||
isRunning: countdown > 0,
|
||||
start,
|
||||
reset,
|
||||
}
|
||||
}
|
||||
+20
-83
@@ -1,84 +1,20 @@
|
||||
import React, { useState, useEffect, useRef } from "react"
|
||||
import { Modal, Tabs, Form, Input, Button, message } from "antd"
|
||||
import { sendVerificationCode, bindContact, type BindContactResponse } from "@/api/auth"
|
||||
|
||||
interface BindContactModalProps {
|
||||
open: boolean
|
||||
onSuccess?: (user: BindContactResponse["user"]) => void
|
||||
onCancel?: () => void
|
||||
}
|
||||
import React from "react"
|
||||
import { Modal, Tabs, Form, Input, Button } from "antd"
|
||||
import { useBindContactForm } from "./hooks/useBindContactForm"
|
||||
import type { BindContactModalProps } from "./types"
|
||||
|
||||
const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, onCancel }) => {
|
||||
const [activeTab, setActiveTab] = useState<"email" | "phone">("email")
|
||||
const [form] = Form.useForm()
|
||||
const [loading, setLoading] = useState(false)
|
||||
const [codeLoading, setCodeLoading] = useState(false)
|
||||
const [countdown, setCountdown] = useState(0)
|
||||
const timerRef = useRef<ReturnType<typeof setInterval> | null>(null)
|
||||
|
||||
useEffect(() => {
|
||||
if (countdown > 0) {
|
||||
timerRef.current = setInterval(() => {
|
||||
setCountdown((prev) => prev - 1)
|
||||
}, 1000)
|
||||
} else if (timerRef.current) {
|
||||
clearInterval(timerRef.current)
|
||||
timerRef.current = null
|
||||
}
|
||||
return () => {
|
||||
if (timerRef.current) clearInterval(timerRef.current)
|
||||
}
|
||||
}, [countdown])
|
||||
|
||||
useEffect(() => {
|
||||
if (open) {
|
||||
form.resetFields()
|
||||
setCountdown(0)
|
||||
}
|
||||
}, [open, form])
|
||||
|
||||
const handleSendCode = async () => {
|
||||
try {
|
||||
const value = form.getFieldValue(activeTab === "email" ? "email" : "phone")
|
||||
if (!value) {
|
||||
message.warning(activeTab === "email" ? "请输入邮箱" : "请输入手机号")
|
||||
return
|
||||
}
|
||||
setCodeLoading(true)
|
||||
await sendVerificationCode({
|
||||
target: activeTab,
|
||||
value,
|
||||
purpose: "bind",
|
||||
})
|
||||
message.success("验证码已发送")
|
||||
setCountdown(60)
|
||||
} catch (error) {
|
||||
// error handled by interceptor
|
||||
} finally {
|
||||
setCodeLoading(false)
|
||||
}
|
||||
}
|
||||
|
||||
const handleSubmit = async () => {
|
||||
try {
|
||||
const values = await form.validateFields()
|
||||
setLoading(true)
|
||||
|
||||
const payload =
|
||||
activeTab === "email"
|
||||
? { email: values.email, email_code: values.code }
|
||||
: { phone: values.phone, phone_code: values.code }
|
||||
|
||||
const result = await bindContact(payload)
|
||||
|
||||
message.success("绑定成功")
|
||||
onSuccess?.(result.user)
|
||||
} catch (error) {
|
||||
// error handled by interceptor
|
||||
} finally {
|
||||
setLoading(false)
|
||||
}
|
||||
}
|
||||
const {
|
||||
activeTab,
|
||||
setActiveTab,
|
||||
form,
|
||||
loading,
|
||||
codeLoading,
|
||||
countdown,
|
||||
codeDisabled,
|
||||
handleSendCode,
|
||||
handleSubmit,
|
||||
} = useBindContactForm({ open, onSuccess, onCancel })
|
||||
|
||||
return (
|
||||
<Modal
|
||||
@@ -111,7 +47,7 @@ const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, on
|
||||
<Input placeholder="请输入邮箱地址" size="large" />
|
||||
</Form.Item>
|
||||
<Form.Item
|
||||
name="code"
|
||||
name="email_code"
|
||||
label="验证码"
|
||||
rules={[{ required: true, message: "请输入验证码" }]}
|
||||
>
|
||||
@@ -121,7 +57,7 @@ const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, on
|
||||
size="large"
|
||||
onClick={handleSendCode}
|
||||
loading={codeLoading}
|
||||
disabled={countdown > 0}
|
||||
disabled={codeDisabled}
|
||||
>
|
||||
{countdown > 0 ? `${countdown}s 后重发` : "发送验证码"}
|
||||
</Button>
|
||||
@@ -146,7 +82,7 @@ const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, on
|
||||
<Input placeholder="请输入手机号" size="large" />
|
||||
</Form.Item>
|
||||
<Form.Item
|
||||
name="code"
|
||||
name="phone_code"
|
||||
label="验证码"
|
||||
rules={[{ required: true, message: "请输入验证码" }]}
|
||||
>
|
||||
@@ -156,7 +92,7 @@ const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, on
|
||||
size="large"
|
||||
onClick={handleSendCode}
|
||||
loading={codeLoading}
|
||||
disabled={countdown > 0}
|
||||
disabled={codeDisabled}
|
||||
>
|
||||
{countdown > 0 ? `${countdown}s 后重发` : "发送验证码"}
|
||||
</Button>
|
||||
@@ -178,3 +114,4 @@ const BindContactModal: React.FC<BindContactModalProps> = ({ open, onSuccess, on
|
||||
}
|
||||
|
||||
export default BindContactModal
|
||||
export type { BindContactModalProps } from "./types"
|
||||
@@ -0,0 +1,14 @@
|
||||
import type { BindContactResponse } from "@/api/auth"
|
||||
|
||||
/** BindContactModal 组件属性 */
|
||||
export interface BindContactModalProps {
|
||||
/** 是否打开 */
|
||||
open: boolean
|
||||
/** 绑定成功回调 */
|
||||
onSuccess?: (user: BindContactResponse["user"]) => void
|
||||
/** 取消回调 */
|
||||
onCancel?: () => void
|
||||
}
|
||||
|
||||
/** 联系方式类型 */
|
||||
export type ContactType = "email" | "phone"
|
||||
@@ -0,0 +1,39 @@
|
||||
import React from "react"
|
||||
import type { StickerItem } from "@/pages/editing-planner/types"
|
||||
import { TEXT_PRESET_STYLES } from "@/pages/editing-planner/constants/sticker"
|
||||
|
||||
interface StickerPreviewProps {
|
||||
sticker: StickerItem
|
||||
}
|
||||
|
||||
export const StickerPreview: React.FC<StickerPreviewProps> = ({ sticker }) => (
|
||||
<div className="sticker-preview-box">
|
||||
<div
|
||||
className="sticker-preview-item"
|
||||
style={{
|
||||
left: `${sticker.x}%`,
|
||||
top: `${sticker.y}%`,
|
||||
width: `${sticker.width}%`,
|
||||
height: `${sticker.height}%`,
|
||||
transform: `translate(-50%, -50%) rotate(${sticker.rotation}deg)`,
|
||||
opacity: sticker.opacity / 100,
|
||||
fontSize: sticker.type === "text" ? `${sticker.font_size}px` : undefined,
|
||||
...TEXT_PRESET_STYLES[sticker.text_preset],
|
||||
}}
|
||||
>
|
||||
{sticker.type === "emoji" && sticker.content}
|
||||
{sticker.type === "text" && sticker.content}
|
||||
{sticker.type === "image" && (
|
||||
<img
|
||||
src={sticker.content}
|
||||
alt="sticker"
|
||||
style={{
|
||||
width: "100%",
|
||||
height: "100%",
|
||||
objectFit: "contain",
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
Regular → Executable
+5
-75
@@ -2,11 +2,9 @@
|
||||
* 选中贴纸的属性编辑器
|
||||
*/
|
||||
import React from "react"
|
||||
import type { StickerItem, TextStickerPreset } from "@/pages/editing-planner/types"
|
||||
import {
|
||||
TEXT_PRESET_STYLES,
|
||||
TEXT_STICKER_PRESET_LABELS,
|
||||
} from "@/pages/editing-planner/constants/sticker"
|
||||
import type { StickerItem } from "@/pages/editing-planner/types"
|
||||
import { StickerPreview } from "./StickerPreview"
|
||||
import { TextStickerPropsEditor } from "./TextStickerPropsEditor"
|
||||
|
||||
interface StickerPropsEditorProps {
|
||||
sticker: StickerItem
|
||||
@@ -121,78 +119,10 @@ const StickerPropsEditor: React.FC<StickerPropsEditorProps> = ({
|
||||
</div>
|
||||
|
||||
{/* 文字贴纸特有属性 */}
|
||||
{sticker.type === "text" && (
|
||||
<>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">花字</span>
|
||||
<select
|
||||
className="sticker-prop-select"
|
||||
value={sticker.text_preset}
|
||||
onChange={(e) =>
|
||||
onUpdate(sticker.id, { text_preset: e.target.value as TextStickerPreset })
|
||||
}
|
||||
>
|
||||
{(Object.keys(TEXT_STICKER_PRESET_LABELS) as TextStickerPreset[]).map((p) => (
|
||||
<option key={p} value={p}>
|
||||
{TEXT_STICKER_PRESET_LABELS[p]}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
</div>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">字号</span>
|
||||
<input
|
||||
type="range"
|
||||
className="sticker-prop-slider"
|
||||
min={12}
|
||||
max={72}
|
||||
value={sticker.font_size}
|
||||
onChange={(e) => onUpdate(sticker.id, { font_size: Number(e.target.value) })}
|
||||
/>
|
||||
<span className="sticker-prop-value">{sticker.font_size}px</span>
|
||||
</div>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">颜色</span>
|
||||
<input
|
||||
type="color"
|
||||
className="sticker-prop-color"
|
||||
value={sticker.text_color}
|
||||
onChange={(e) => onUpdate(sticker.id, { text_color: e.target.value })}
|
||||
/>
|
||||
</div>
|
||||
</>
|
||||
)}
|
||||
<TextStickerPropsEditor sticker={sticker} onUpdate={onUpdate} />
|
||||
|
||||
{/* 预览 */}
|
||||
<div className="sticker-preview-box">
|
||||
<div
|
||||
className="sticker-preview-item"
|
||||
style={{
|
||||
left: `${sticker.x}%`,
|
||||
top: `${sticker.y}%`,
|
||||
width: `${sticker.width}%`,
|
||||
height: `${sticker.width}%`,
|
||||
transform: `translate(-50%, -50%) rotate(${sticker.rotation}deg)`,
|
||||
opacity: sticker.opacity / 100,
|
||||
fontSize: sticker.type === "text" ? `${sticker.font_size}px` : undefined,
|
||||
...TEXT_PRESET_STYLES[sticker.text_preset],
|
||||
}}
|
||||
>
|
||||
{sticker.type === "emoji" && sticker.content}
|
||||
{sticker.type === "text" && sticker.content}
|
||||
{sticker.type === "image" && (
|
||||
<img
|
||||
src={sticker.content}
|
||||
alt="sticker"
|
||||
style={{
|
||||
width: "100%",
|
||||
height: "100%",
|
||||
objectFit: "contain",
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
<StickerPreview sticker={sticker} />
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
+57
@@ -0,0 +1,57 @@
|
||||
import React from "react"
|
||||
import type { StickerItem, TextStickerPreset } from "@/pages/editing-planner/types"
|
||||
import { TEXT_STICKER_PRESET_LABELS } from "@/pages/editing-planner/constants/sticker"
|
||||
|
||||
interface TextStickerPropsEditorProps {
|
||||
sticker: StickerItem
|
||||
onUpdate: (id: string, partial: Partial<StickerItem>) => void
|
||||
}
|
||||
|
||||
export const TextStickerPropsEditor: React.FC<TextStickerPropsEditorProps> = ({
|
||||
sticker,
|
||||
onUpdate,
|
||||
}) => {
|
||||
if (sticker.type !== "text") return null
|
||||
|
||||
return (
|
||||
<>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">花字</span>
|
||||
<select
|
||||
className="sticker-prop-select"
|
||||
value={sticker.text_preset}
|
||||
onChange={(e) =>
|
||||
onUpdate(sticker.id, { text_preset: e.target.value as TextStickerPreset })
|
||||
}
|
||||
>
|
||||
{(Object.keys(TEXT_STICKER_PRESET_LABELS) as TextStickerPreset[]).map((p) => (
|
||||
<option key={p} value={p}>
|
||||
{TEXT_STICKER_PRESET_LABELS[p]}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
</div>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">字号</span>
|
||||
<input
|
||||
type="range"
|
||||
className="sticker-prop-slider"
|
||||
min={12}
|
||||
max={72}
|
||||
value={sticker.font_size}
|
||||
onChange={(e) => onUpdate(sticker.id, { font_size: Number(e.target.value) })}
|
||||
/>
|
||||
<span className="sticker-prop-value">{sticker.font_size}px</span>
|
||||
</div>
|
||||
<div className="sticker-prop-row">
|
||||
<span className="sticker-prop-label">颜色</span>
|
||||
<input
|
||||
type="color"
|
||||
className="sticker-prop-color"
|
||||
value={sticker.text_color}
|
||||
onChange={(e) => onUpdate(sticker.id, { text_color: e.target.value })}
|
||||
/>
|
||||
</div>
|
||||
</>
|
||||
)
|
||||
}
|
||||
@@ -1,3 +1,7 @@
|
||||
/**
|
||||
* Auth API 测试
|
||||
* 对应 api/auth/ 目录化后的模块
|
||||
*/
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest"
|
||||
import {
|
||||
normalizeUser,
|
||||
@@ -10,7 +14,6 @@ import {
|
||||
resetPassword,
|
||||
verifyEmail,
|
||||
} from "@/api/auth"
|
||||
|
||||
const mockPost = vi.fn()
|
||||
const mockGet = vi.fn()
|
||||
const mockAxiosPost = vi.fn()
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import { describe, expect, it } from "vitest"
|
||||
import BindContactModal from "@/components/auth/BindContactModal"
|
||||
import type { BindContactModalProps, ContactType } from "@/components/auth/BindContactModal/types"
|
||||
import { useBindContactForm } from "@/components/auth/BindContactModal/hooks/useBindContactForm"
|
||||
import { useCountdown } from "@/components/auth/BindContactModal/hooks/useCountdown"
|
||||
|
||||
describe("BindContactModal", () => {
|
||||
it("主组件可正常导入", () => {
|
||||
expect(BindContactModal).toBeDefined()
|
||||
})
|
||||
|
||||
it("Hook 可正常导入", () => {
|
||||
expect(useBindContactForm).toBeDefined()
|
||||
expect(useCountdown).toBeDefined()
|
||||
})
|
||||
|
||||
it("类型导出正常", () => {
|
||||
const props: BindContactModalProps = { open: false }
|
||||
expect(props.open).toBe(false)
|
||||
|
||||
const contactType: ContactType = "email"
|
||||
expect(contactType).toBe("email")
|
||||
})
|
||||
})
|
||||
Regular → Executable
+2
@@ -61,6 +61,8 @@ import "@/pages/editing-planner/components/pip-config/LayerConfig"
|
||||
import "@/pages/editing-planner/components/sticker/StickerLibrary"
|
||||
import "@/pages/editing-planner/components/sticker/StickerList"
|
||||
import "@/pages/editing-planner/components/sticker/StickerPropsEditor"
|
||||
import "@/pages/editing-planner/components/sticker/StickerPreview"
|
||||
import "@/pages/editing-planner/components/sticker/TextStickerPropsEditor"
|
||||
import "@/pages/editing-planner/components/filter/FilterPresetGrid"
|
||||
import "@/pages/editing-planner/components/filter/FilterManualAdjust"
|
||||
import "@/pages/editing-planner/components/intro-outro/IntroOutroBlock"
|
||||
|
||||
Executable
+281
@@ -0,0 +1,281 @@
|
||||
"""edit_template 剪辑模板实体单测."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from domain.edit_template import EditTemplate, EditTemplateStatus
|
||||
from domain.editing_mode import EditingMode
|
||||
|
||||
# ── EditTemplateStatus 枚举 ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditTemplateStatus:
|
||||
"""EditTemplateStatus 枚举"""
|
||||
|
||||
def test_enum_values(self):
|
||||
assert EditTemplateStatus.ACTIVE.value == "active"
|
||||
assert EditTemplateStatus.INACTIVE.value == "inactive"
|
||||
|
||||
def test_is_str_enum(self):
|
||||
assert isinstance(EditTemplateStatus.ACTIVE, str)
|
||||
assert EditTemplateStatus.ACTIVE == "active"
|
||||
|
||||
def test_from_string(self):
|
||||
assert EditTemplateStatus("active") == EditTemplateStatus.ACTIVE
|
||||
assert EditTemplateStatus("inactive") == EditTemplateStatus.INACTIVE
|
||||
|
||||
def test_invalid_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
EditTemplateStatus("deleted")
|
||||
|
||||
|
||||
# ── EditTemplate.create 工厂方法 ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditTemplateCreate:
|
||||
"""EditTemplate.create 工厂方法"""
|
||||
|
||||
def test_minimal_create(self):
|
||||
t = EditTemplate.create("测试模板")
|
||||
assert t.id is not None
|
||||
assert len(t.id) == 32 # uuid4 hex
|
||||
assert t.name == "测试模板"
|
||||
assert t.description == ""
|
||||
assert t.template_type == "default"
|
||||
assert t.editing_mode == "one_take"
|
||||
assert t.config == {}
|
||||
assert t.preview_url == ""
|
||||
assert t.sort_weight == 0
|
||||
assert t.status == EditTemplateStatus.ACTIVE
|
||||
assert t.version == 1
|
||||
|
||||
def test_unique_ids(self):
|
||||
t1 = EditTemplate.create("模板A")
|
||||
t2 = EditTemplate.create("模板B")
|
||||
assert t1.id != t2.id
|
||||
|
||||
def test_custom_fields(self):
|
||||
t = EditTemplate.create(
|
||||
"自定义模板",
|
||||
description="这是一个自定义模板",
|
||||
template_type="story",
|
||||
editing_mode="one_take",
|
||||
config={"key": "value"},
|
||||
preview_url="https://example.com/preview.mp4",
|
||||
sort_weight=100,
|
||||
status=EditTemplateStatus.INACTIVE,
|
||||
version=2,
|
||||
)
|
||||
assert t.name == "自定义模板"
|
||||
assert t.description == "这是一个自定义模板"
|
||||
assert t.template_type == "story"
|
||||
assert t.editing_mode == "one_take"
|
||||
assert t.config == {"key": "value"}
|
||||
assert t.preview_url == "https://example.com/preview.mp4"
|
||||
assert t.sort_weight == 100
|
||||
assert t.status == EditTemplateStatus.INACTIVE
|
||||
assert t.version == 2
|
||||
|
||||
def test_name_stripped(self):
|
||||
t = EditTemplate.create(" 带空格的模板 ")
|
||||
assert t.name == "带空格的模板"
|
||||
|
||||
def test_empty_name_raises(self):
|
||||
with pytest.raises(ValueError, match="名称"):
|
||||
EditTemplate.create("")
|
||||
|
||||
def test_whitespace_only_name_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
EditTemplate.create(" ")
|
||||
|
||||
def test_invalid_editing_mode_raises(self):
|
||||
with pytest.raises(ValueError, match="editing_mode"):
|
||||
EditTemplate.create("测试", editing_mode="invalid_mode")
|
||||
|
||||
def test_empty_editing_mode_falls_back_to_default(self):
|
||||
t = EditTemplate.create("测试", editing_mode="")
|
||||
assert t.editing_mode == "one_take"
|
||||
|
||||
def test_whitespace_editing_mode_falls_back(self):
|
||||
t = EditTemplate.create("测试", editing_mode=" ")
|
||||
assert t.editing_mode == "one_take"
|
||||
|
||||
def test_editing_mode_stripped(self):
|
||||
t = EditTemplate.create("测试", editing_mode=" one_take ")
|
||||
assert t.editing_mode == "one_take"
|
||||
|
||||
def test_description_stripped(self):
|
||||
t = EditTemplate.create("测试", description=" 描述 ")
|
||||
assert t.description == "描述"
|
||||
|
||||
def test_template_type_stripped(self):
|
||||
t = EditTemplate.create("测试", template_type=" vlog ")
|
||||
assert t.template_type == "vlog"
|
||||
|
||||
def test_empty_template_type_falls_back(self):
|
||||
t = EditTemplate.create("测试", template_type="")
|
||||
assert t.template_type == "default"
|
||||
|
||||
def test_none_config_becomes_empty_dict(self):
|
||||
t = EditTemplate.create("测试", config=None)
|
||||
assert t.config == {}
|
||||
assert isinstance(t.config, dict)
|
||||
|
||||
def test_preview_url_stripped(self):
|
||||
t = EditTemplate.create("测试", preview_url=" https://x.com/a.mp4 ")
|
||||
assert t.preview_url == "https://x.com/a.mp4"
|
||||
|
||||
def test_timestamps_are_utc(self):
|
||||
t = EditTemplate.create("测试")
|
||||
assert t.created_at.tzinfo is not None
|
||||
assert t.updated_at.tzinfo is not None
|
||||
|
||||
def test_created_at_equals_updated_at_on_create(self):
|
||||
t = EditTemplate.create("测试")
|
||||
# 创建时两个时间应该非常接近
|
||||
diff = abs((t.updated_at - t.created_at).total_seconds())
|
||||
assert diff < 1.0
|
||||
|
||||
|
||||
# ── 状态操作 ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditTemplateStatusOperations:
|
||||
"""EditTemplate 状态操作"""
|
||||
|
||||
def test_activate_sets_active(self):
|
||||
t = EditTemplate.create("测试", status=EditTemplateStatus.INACTIVE)
|
||||
t.activate()
|
||||
assert t.status == EditTemplateStatus.ACTIVE
|
||||
assert t.is_active is True
|
||||
|
||||
def test_deactivate_sets_inactive(self):
|
||||
t = EditTemplate.create("测试")
|
||||
t.deactivate()
|
||||
assert t.status == EditTemplateStatus.INACTIVE
|
||||
assert t.is_active is False
|
||||
|
||||
def test_is_active_true(self):
|
||||
t = EditTemplate.create("测试")
|
||||
assert t.is_active is True
|
||||
|
||||
def test_is_active_false(self):
|
||||
t = EditTemplate.create("测试", status=EditTemplateStatus.INACTIVE)
|
||||
assert t.is_active is False
|
||||
|
||||
def test_activate_updates_updated_at(self):
|
||||
t = EditTemplate.create("测试", status=EditTemplateStatus.INACTIVE)
|
||||
old_updated = t.updated_at
|
||||
t.activate()
|
||||
assert t.updated_at >= old_updated
|
||||
|
||||
def test_deactivate_updates_updated_at(self):
|
||||
t = EditTemplate.create("测试")
|
||||
old_updated = t.updated_at
|
||||
t.deactivate()
|
||||
assert t.updated_at >= old_updated
|
||||
|
||||
|
||||
# ── 版本操作 ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditTemplateVersion:
|
||||
"""EditTemplate 版本操作"""
|
||||
|
||||
def test_bump_version_increments(self):
|
||||
t = EditTemplate.create("测试")
|
||||
assert t.version == 1
|
||||
t.bump_version()
|
||||
assert t.version == 2
|
||||
|
||||
def test_bump_version_multiple(self):
|
||||
t = EditTemplate.create("测试", version=5)
|
||||
t.bump_version()
|
||||
t.bump_version()
|
||||
t.bump_version()
|
||||
assert t.version == 8
|
||||
|
||||
def test_bump_version_updates_updated_at(self):
|
||||
t = EditTemplate.create("测试")
|
||||
old_updated = t.updated_at
|
||||
t.bump_version()
|
||||
assert t.updated_at >= old_updated
|
||||
|
||||
|
||||
# ── dataclass 基础特性 ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditTemplateBasics:
|
||||
"""EditTemplate 基础特性"""
|
||||
|
||||
def test_slots_no_extra_attrs(self):
|
||||
t = EditTemplate.create("测试")
|
||||
with pytest.raises(AttributeError):
|
||||
t.nonexistent_field = "value"
|
||||
|
||||
def test_direct_construction_minimal(self):
|
||||
# 最小构造:仅必填字段 + 状态,其余走默认值
|
||||
t = EditTemplate(
|
||||
id="custom_id",
|
||||
name="直接构造",
|
||||
status=EditTemplateStatus.ACTIVE,
|
||||
)
|
||||
assert t.id == "custom_id"
|
||||
assert t.name == "直接构造"
|
||||
assert t.status == EditTemplateStatus.ACTIVE
|
||||
# 默认值检查
|
||||
assert t.description == ""
|
||||
assert t.config == {}
|
||||
assert t.version == 1
|
||||
assert t.editing_mode == EditingMode.ONE_TAKE.value
|
||||
assert isinstance(t.created_at, datetime)
|
||||
assert isinstance(t.updated_at, datetime)
|
||||
|
||||
def test_direct_construction_full(self):
|
||||
# 完整构造:所有字段都传
|
||||
now = datetime(2025, 1, 1, tzinfo=timezone.utc)
|
||||
t = EditTemplate(
|
||||
id="full_id",
|
||||
name="完整构造",
|
||||
description="测试描述",
|
||||
template_type="custom",
|
||||
editing_mode=EditingMode.PIP.value,
|
||||
config={"key": "value"},
|
||||
preview_url="https://example.com/preview.jpg",
|
||||
sort_weight=100,
|
||||
status=EditTemplateStatus.INACTIVE,
|
||||
version=3,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
assert t.id == "full_id"
|
||||
assert t.name == "完整构造"
|
||||
assert t.description == "测试描述"
|
||||
assert t.template_type == "custom"
|
||||
assert t.editing_mode == EditingMode.PIP.value
|
||||
assert t.config == {"key": "value"}
|
||||
assert t.preview_url == "https://example.com/preview.jpg"
|
||||
assert t.sort_weight == 100
|
||||
assert t.status == EditTemplateStatus.INACTIVE
|
||||
assert t.version == 3
|
||||
assert t.created_at == now
|
||||
assert t.updated_at == now
|
||||
|
||||
def test_config_is_independent(self):
|
||||
# 不同实例的 config 应该是独立的 dict
|
||||
t1 = EditTemplate.create("模板1")
|
||||
t2 = EditTemplate.create("模板2")
|
||||
t1.config["key"] = "value"
|
||||
assert "key" not in t2.config
|
||||
|
||||
def test_equality(self):
|
||||
# 两个不同实例即使内容相同也不等(id不同)
|
||||
t1 = EditTemplate.create("同名模板")
|
||||
t2 = EditTemplate.create("同名模板")
|
||||
assert t1 != t2
|
||||
|
||||
def test_same_id_equal(self):
|
||||
now = datetime.now(timezone.utc)
|
||||
t1 = EditTemplate(id="same", name="同名", created_at=now, updated_at=now)
|
||||
t2 = EditTemplate(id="same", name="同名", created_at=now, updated_at=now)
|
||||
assert t1 == t2
|
||||
+298
-387
@@ -1,68 +1,63 @@
|
||||
"""JWT 服务与处理器单元测试."""
|
||||
"""JWT 服务单元测试 — wave130."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import jwt
|
||||
import jwt as pyjwt
|
||||
import pytest
|
||||
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
|
||||
|
||||
from packages.application.auth.jwt_handler import (
|
||||
JWTHandler,
|
||||
configure_jwt_handler,
|
||||
get_jwt_handler,
|
||||
)
|
||||
from packages.application.auth.jwt_service import (
|
||||
JWTConfig,
|
||||
JWTService,
|
||||
TokenType,
|
||||
)
|
||||
|
||||
# ── 测试常量 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
TEST_SECRET = "test-secret-key-for-unit-testing-only-not-for-production"
|
||||
STRONG_SECRET = "x" * 32 # 满足长度要求的测试密钥
|
||||
# ── 测试常量 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
# ── JWTConfig 测试 ───────────────────────────────────────────────────────────
|
||||
TEST_SECRET = "test-secret-key-for-unit-testing-only-1234567890"
|
||||
TEST_ALGORITHM = "HS256"
|
||||
|
||||
|
||||
# ── JWTConfig 配置 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestJWTConfig:
|
||||
"""JWTConfig 配置类测试"""
|
||||
|
||||
def test_init_with_valid_secret(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET)
|
||||
assert config.SECRET_KEY == STRONG_SECRET
|
||||
def test_normal_config(self):
|
||||
config = JWTConfig(secret_key=TEST_SECRET)
|
||||
assert config.SECRET_KEY == TEST_SECRET
|
||||
assert config.ALGORITHM == "HS256"
|
||||
assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 15
|
||||
assert config.REFRESH_TOKEN_EXPIRE_DAYS == 7
|
||||
|
||||
def test_init_custom_values(self):
|
||||
def test_custom_config(self):
|
||||
config = JWTConfig(
|
||||
secret_key=STRONG_SECRET,
|
||||
secret_key=TEST_SECRET,
|
||||
algorithm="HS384",
|
||||
access_token_expire_minutes=60,
|
||||
refresh_token_expire_days=14,
|
||||
refresh_token_expire_days=30,
|
||||
)
|
||||
assert config.SECRET_KEY == STRONG_SECRET
|
||||
assert config.ALGORITHM == "HS384"
|
||||
assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 60
|
||||
assert config.REFRESH_TOKEN_EXPIRE_DAYS == 14
|
||||
assert config.REFRESH_TOKEN_EXPIRE_DAYS == 30
|
||||
|
||||
def test_empty_secret_raises(self):
|
||||
with pytest.raises(ValueError, match="secret_key must be provided"):
|
||||
JWTConfig(secret_key="")
|
||||
|
||||
def test_whitespace_only_secret_raises(self):
|
||||
with pytest.raises(ValueError, match="secret_key must be provided"):
|
||||
def test_whitespace_secret_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
JWTConfig(secret_key=" ")
|
||||
|
||||
def test_none_secret_raises(self):
|
||||
with pytest.raises(ValueError, match="secret_key must be provided"):
|
||||
JWTConfig(secret_key=None)
|
||||
with pytest.raises(ValueError):
|
||||
JWTConfig(secret_key=None) # type: ignore
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"insecure_secret",
|
||||
"bad_secret",
|
||||
[
|
||||
"your-secret-key-change-in-production",
|
||||
"your-secret-key",
|
||||
@@ -73,407 +68,323 @@ class TestJWTConfig:
|
||||
"Your-Secret-Key",
|
||||
],
|
||||
)
|
||||
def test_insecure_default_secret_raises(self, insecure_secret):
|
||||
def test_insecure_defaults_rejected(self, bad_secret):
|
||||
with pytest.raises(ValueError, match="insecure"):
|
||||
JWTConfig(secret_key=insecure_secret)
|
||||
|
||||
def test_zero_expire_minutes_allowed(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
|
||||
assert config.ACCESS_TOKEN_EXPIRE_MINUTES == 0
|
||||
|
||||
def test_negative_expire_days_allowed(self):
|
||||
# 配置类不校验合理性,由业务层判断
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=-1)
|
||||
assert config.REFRESH_TOKEN_EXPIRE_DAYS == -1
|
||||
JWTConfig(secret_key=bad_secret)
|
||||
|
||||
|
||||
# ── JWTService 初始化测试 ────────────────────────────────────────────────────
|
||||
# ── JWTService 初始化 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestJWTServiceInit:
|
||||
"""JWTService 初始化测试"""
|
||||
|
||||
def test_init_with_config(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET)
|
||||
def test_with_config_works(self):
|
||||
config = JWTConfig(secret_key=TEST_SECRET)
|
||||
service = JWTService(config)
|
||||
assert service.config is config
|
||||
|
||||
def test_init_none_config_raises(self):
|
||||
with pytest.raises(ValueError, match="JWTService requires a JWTConfig"):
|
||||
def test_none_config_raises(self):
|
||||
with pytest.raises(ValueError, match="JWTService requires"):
|
||||
JWTService(None)
|
||||
|
||||
|
||||
# ── TokenType 测试 ───────────────────────────────────────────────────────────
|
||||
# ── create_access_token ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreateAccessToken:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_creates_valid_jwt(self):
|
||||
token = self.service.create_access_token(user_id="user123")
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
# JWT 格式:xxx.yyy.zzz
|
||||
assert token.count(".") == 2
|
||||
|
||||
def test_payload_contains_user_id(self):
|
||||
token = self.service.create_access_token(user_id="user_001")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "user_001"
|
||||
|
||||
def test_payload_contains_role(self):
|
||||
token = self.service.create_access_token(user_id="u1", role="admin")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["role"] == "admin"
|
||||
|
||||
def test_default_role_empty(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["role"] == ""
|
||||
|
||||
def test_token_type_is_access(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["type"] == TokenType.ACCESS
|
||||
|
||||
def test_has_iat_and_exp(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert "iat" in payload
|
||||
assert "exp" in payload
|
||||
assert payload["exp"] > payload["iat"]
|
||||
|
||||
def test_expiration_correct(self):
|
||||
"""过期时间大约等于当前时间 + 配置的分钟数."""
|
||||
config = JWTConfig(secret_key=TEST_SECRET, access_token_expire_minutes=30)
|
||||
service = JWTService(config)
|
||||
before = datetime.now(timezone.utc)
|
||||
token = service.create_access_token(user_id="u1")
|
||||
after = datetime.now(timezone.utc)
|
||||
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
exp = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
|
||||
|
||||
min_expected = before + timedelta(minutes=30) - timedelta(seconds=1)
|
||||
max_expected = after + timedelta(minutes=30) + timedelta(seconds=1)
|
||||
assert min_expected <= exp <= max_expected
|
||||
|
||||
def test_additional_claims_included(self):
|
||||
extra = {"email": "test@example.com", "org_id": "org_001", "level": 5}
|
||||
token = self.service.create_access_token(user_id="u1", additional_claims=extra)
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["email"] == "test@example.com"
|
||||
assert payload["org_id"] == "org_001"
|
||||
assert payload["level"] == 5
|
||||
|
||||
def test_additional_claims_none(self):
|
||||
token = self.service.create_access_token(user_id="u1", additional_claims=None)
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert "email" not in payload
|
||||
|
||||
def test_signed_with_correct_key(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
# 用正确的密钥可以解码
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "u1"
|
||||
# 用错误的密钥无法解码
|
||||
with pytest.raises(InvalidTokenError):
|
||||
pyjwt.decode(token, "wrong-secret", algorithms=["HS256"])
|
||||
|
||||
|
||||
# ── create_refresh_token ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreateRefreshToken:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_creates_valid_token(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="sess_001")
|
||||
assert isinstance(token, str)
|
||||
assert token.count(".") == 2
|
||||
|
||||
def test_payload_contains_session_id(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="sess_abc")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["session_id"] == "sess_abc"
|
||||
assert payload["sub"] == "u1"
|
||||
|
||||
def test_token_type_is_refresh(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
assert payload["type"] == TokenType.REFRESH
|
||||
|
||||
def test_refresh_expiration_days(self):
|
||||
config = JWTConfig(secret_key=TEST_SECRET, refresh_token_expire_days=7)
|
||||
service = JWTService(config)
|
||||
before = datetime.now(timezone.utc)
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
after = datetime.now(timezone.utc)
|
||||
|
||||
payload = pyjwt.decode(token, TEST_SECRET, algorithms=["HS256"])
|
||||
exp = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
|
||||
|
||||
min_exp = before + timedelta(days=7) - timedelta(seconds=1)
|
||||
max_exp = after + timedelta(days=7, seconds=1)
|
||||
assert min_exp <= exp <= max_exp
|
||||
|
||||
|
||||
# ── verify_token ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyToken:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_valid_token_returns_payload(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
payload = self.service.verify_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
|
||||
def test_expired_token_raises(self):
|
||||
# 创建一个 1 秒过期的 token
|
||||
config = JWTConfig(secret_key=TEST_SECRET, access_token_expire_minutes=1)
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="u1")
|
||||
|
||||
# 等待过期(用 pyjwt 直接构造过期 token 更可靠)
|
||||
expired_payload = {
|
||||
"sub": "u1",
|
||||
"type": "access",
|
||||
"exp": datetime.now(timezone.utc) - timedelta(seconds=10),
|
||||
}
|
||||
expired_token = pyjwt.encode(expired_payload, TEST_SECRET, algorithm="HS256")
|
||||
|
||||
with pytest.raises(ExpiredSignatureError, match="expired"):
|
||||
self.service.verify_token(expired_token)
|
||||
|
||||
def test_invalid_token_raises(self):
|
||||
with pytest.raises(InvalidTokenError, match="Invalid token"):
|
||||
self.service.verify_token("not-a-valid-jwt-token")
|
||||
|
||||
def test_wrong_signature_raises(self):
|
||||
token = pyjwt.encode({"sub": "u1"}, "different-secret", algorithm="HS256")
|
||||
with pytest.raises(InvalidTokenError):
|
||||
self.service.verify_token(token)
|
||||
|
||||
def test_tampered_payload_raises(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
# 尝试篡改:JWT 有签名保护,篡改会导致验证失败
|
||||
parts = token.split(".")
|
||||
assert len(parts) == 3
|
||||
# 把 payload 部分替换(不会成功,因为签名不对)
|
||||
import base64
|
||||
|
||||
fake_payload = base64.urlsafe_b64encode(b'{"sub":"admin","role":"admin"}').rstrip(b"=").decode()
|
||||
tampered = f"{parts[0]}.{fake_payload}.{parts[2]}"
|
||||
with pytest.raises(InvalidTokenError):
|
||||
self.service.verify_token(tampered)
|
||||
|
||||
|
||||
# ── verify_access_token ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyAccessToken:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_access_token_passes(self):
|
||||
token = self.service.create_access_token(user_id="u1", role="user")
|
||||
payload = self.service.verify_access_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["type"] == "access"
|
||||
|
||||
def test_refresh_token_rejected(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
with pytest.raises(ValueError, match="Token type must be 'access'"):
|
||||
self.service.verify_access_token(token)
|
||||
|
||||
def test_expired_token_raises(self):
|
||||
expired_payload = {
|
||||
"sub": "u1",
|
||||
"type": "access",
|
||||
"exp": datetime.now(timezone.utc) - timedelta(seconds=10),
|
||||
}
|
||||
token = pyjwt.encode(expired_payload, TEST_SECRET, algorithm="HS256")
|
||||
with pytest.raises(ExpiredSignatureError):
|
||||
self.service.verify_access_token(token)
|
||||
|
||||
|
||||
# ── verify_refresh_token ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyRefreshToken:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_refresh_token_passes(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="sess_001")
|
||||
payload = self.service.verify_refresh_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["session_id"] == "sess_001"
|
||||
|
||||
def test_access_token_rejected(self):
|
||||
token = self.service.create_access_token(user_id="u1")
|
||||
with pytest.raises(ValueError, match="Token type must be 'refresh'"):
|
||||
self.service.verify_refresh_token(token)
|
||||
|
||||
def test_has_session_id(self):
|
||||
token = self.service.create_refresh_token(user_id="u1", session_id="custom_sess")
|
||||
payload = self.service.verify_refresh_token(token)
|
||||
assert payload["session_id"] == "custom_sess"
|
||||
|
||||
|
||||
# ── TokenType 常量 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTokenType:
|
||||
"""TokenType 常量测试"""
|
||||
|
||||
def test_access_value(self):
|
||||
assert TokenType.ACCESS == "access"
|
||||
|
||||
def test_refresh_value(self):
|
||||
assert TokenType.REFRESH == "refresh"
|
||||
|
||||
def test_access_and_refresh_different(self):
|
||||
def test_different_types(self):
|
||||
assert TokenType.ACCESS != TokenType.REFRESH
|
||||
|
||||
|
||||
# ── JWTService create_access_token 测试 ─────────────────────────────────────
|
||||
# ── 多算法支持 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreateAccessToken:
|
||||
"""创建 access_token 测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return JWTService(JWTConfig(secret_key=STRONG_SECRET))
|
||||
|
||||
def test_creates_valid_jwt_string(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
def test_token_contains_user_id_as_sub(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "user-123"
|
||||
|
||||
def test_token_type_is_access(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["type"] == TokenType.ACCESS
|
||||
|
||||
def test_default_role_is_empty_string(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["role"] == ""
|
||||
|
||||
def test_custom_role(self, service):
|
||||
token = service.create_access_token(user_id="user-123", role="admin")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["role"] == "admin"
|
||||
|
||||
def test_has_iat_and_exp(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert "iat" in payload
|
||||
assert "exp" in payload
|
||||
assert payload["exp"] > payload["iat"]
|
||||
|
||||
def test_expire_matches_config(self, service):
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
iat = datetime.fromtimestamp(payload["iat"], tz=timezone.utc)
|
||||
exp = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
|
||||
delta = exp - iat
|
||||
assert delta.total_seconds() == 15 * 60 # 15分钟
|
||||
|
||||
def test_custom_expire_time(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=30)
|
||||
class TestDifferentAlgorithms:
|
||||
def test_hs384_works(self):
|
||||
config = JWTConfig(secret_key=TEST_SECRET * 2, algorithm="HS384")
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="user-123")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
delta = payload["exp"] - payload["iat"]
|
||||
assert delta == 30 * 60
|
||||
|
||||
def test_additional_claims(self, service):
|
||||
extra = {"custom_field": "value", "another": 42}
|
||||
token = service.create_access_token(user_id="user-123", additional_claims=extra)
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["custom_field"] == "value"
|
||||
assert payload["another"] == 42
|
||||
|
||||
def test_additional_claims_can_override_standard(self, service):
|
||||
# additional_claims 可以覆盖标准字段(由调用者负责)
|
||||
token = service.create_access_token(
|
||||
user_id="user-123",
|
||||
additional_claims={"sub": "overridden"},
|
||||
)
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "overridden"
|
||||
|
||||
def test_additional_claims_none_is_same_as_empty(self, service):
|
||||
token = service.create_access_token(user_id="user-123", additional_claims=None)
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "user-123"
|
||||
|
||||
def test_uses_correct_algorithm(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, algorithm="HS384")
|
||||
service = JWTService(config)
|
||||
token = service.create_access_token(user_id="u1")
|
||||
# 用 HS256 解码应该失败
|
||||
with pytest.raises(InvalidTokenError):
|
||||
jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
# 用 HS384 解码应该成功
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS384"])
|
||||
assert payload["sub"] == "u1"
|
||||
|
||||
|
||||
# ── JWTService create_refresh_token 测试 ────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreateRefreshToken:
|
||||
"""创建 refresh_token 测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return JWTService(JWTConfig(secret_key=STRONG_SECRET))
|
||||
|
||||
def test_creates_valid_string(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
def test_contains_user_id_and_session_id(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="sess-abc")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["session_id"] == "sess-abc"
|
||||
|
||||
def test_token_type_is_refresh(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert payload["type"] == TokenType.REFRESH
|
||||
|
||||
def test_has_iat_and_exp(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
assert "iat" in payload
|
||||
assert "exp" in payload
|
||||
assert payload["exp"] > payload["iat"]
|
||||
|
||||
def test_expire_matches_config_days(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
delta = payload["exp"] - payload["iat"]
|
||||
assert delta == 7 * 24 * 60 * 60 # 7天
|
||||
|
||||
def test_custom_refresh_expire_days(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=30)
|
||||
service = JWTService(config)
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = jwt.decode(token, STRONG_SECRET, algorithms=["HS256"])
|
||||
delta = payload["exp"] - payload["iat"]
|
||||
assert delta == 30 * 24 * 60 * 60
|
||||
|
||||
|
||||
# ── JWTService verify_token 测试 ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyToken:
|
||||
"""通用 Token 验证测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return JWTService(JWTConfig(secret_key=STRONG_SECRET))
|
||||
|
||||
def test_verify_valid_access_token(self, service):
|
||||
token = service.create_access_token(user_id="u1")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["type"] == TokenType.ACCESS
|
||||
|
||||
def test_verify_valid_refresh_token(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["session_id"] == "s1"
|
||||
|
||||
def test_verify_expired_token_raises(self, service):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
|
||||
svc = JWTService(config)
|
||||
token = svc.create_access_token(user_id="u1")
|
||||
# 0 分钟过期,立即过期
|
||||
time.sleep(0.1) # 稍微等一下确保过期
|
||||
with pytest.raises(ExpiredSignatureError):
|
||||
svc.verify_token(token)
|
||||
|
||||
def test_verify_wrong_secret_raises(self, service):
|
||||
token = service.create_access_token(user_id="u1")
|
||||
other_service = JWTService(JWTConfig(secret_key="different-secret-1234567890"))
|
||||
with pytest.raises(InvalidTokenError):
|
||||
other_service.verify_token(token)
|
||||
|
||||
def test_verify_tampered_token_raises(self, service):
|
||||
token = service.create_access_token(user_id="u1")
|
||||
# 篡改 token 中间部分
|
||||
parts = token.split(".")
|
||||
assert len(parts) == 3
|
||||
tampered = parts[0] + "." + parts[1][:-1] + "A." + parts[2]
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token(tampered)
|
||||
|
||||
def test_verify_empty_string_raises(self, service):
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token("")
|
||||
|
||||
def test_verify_garbage_string_raises(self, service):
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service.verify_token("not.a.valid.jwt.token")
|
||||
|
||||
def test_verify_returns_dict(self, service):
|
||||
token = service.create_access_token(user_id="u1", role="admin")
|
||||
payload = service.verify_token(token)
|
||||
assert isinstance(payload, dict)
|
||||
assert "sub" in payload
|
||||
assert "role" in payload
|
||||
|
||||
|
||||
# ── JWTService verify_access_token 测试 ─────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyAccessToken:
|
||||
"""Access Token 专属验证测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return JWTService(JWTConfig(secret_key=STRONG_SECRET))
|
||||
|
||||
def test_valid_access_token_passes(self, service):
|
||||
token = service.create_access_token(user_id="u1", role="admin")
|
||||
payload = service.verify_access_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["role"] == "admin"
|
||||
|
||||
def test_refresh_token_fails_type_check(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
with pytest.raises(ValueError, match="Token type must be 'access'"):
|
||||
service.verify_access_token(token)
|
||||
|
||||
def test_token_without_type_field_raises(self, service):
|
||||
# 手动构造一个没有 type 字段的 token
|
||||
payload_data = {"sub": "u1", "iat": 1000, "exp": 9999999999}
|
||||
token = jwt.encode(payload_data, STRONG_SECRET, algorithm="HS256")
|
||||
with pytest.raises(ValueError, match="Token type must be 'access'"):
|
||||
service.verify_access_token(token)
|
||||
|
||||
def test_expired_access_token_raises_expired_error(self, service):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, access_token_expire_minutes=0)
|
||||
svc = JWTService(config)
|
||||
token = svc.create_access_token(user_id="u1")
|
||||
time.sleep(0.1)
|
||||
with pytest.raises(ExpiredSignatureError):
|
||||
svc.verify_access_token(token)
|
||||
|
||||
|
||||
# ── JWTService verify_refresh_token 测试 ────────────────────────────────────
|
||||
|
||||
|
||||
class TestVerifyRefreshToken:
|
||||
"""Refresh Token 专属验证测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return JWTService(JWTConfig(secret_key=STRONG_SECRET))
|
||||
|
||||
def test_valid_refresh_token_passes(self, service):
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
payload = service.verify_refresh_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["session_id"] == "s1"
|
||||
|
||||
def test_access_token_fails_type_check(self, service):
|
||||
token = service.create_access_token(user_id="u1")
|
||||
with pytest.raises(ValueError, match="Token type must be 'refresh'"):
|
||||
service.verify_refresh_token(token)
|
||||
|
||||
def test_token_without_type_field_raises(self, service):
|
||||
payload_data = {"sub": "u1", "session_id": "s1", "iat": 1000, "exp": 9999999999}
|
||||
token = jwt.encode(payload_data, STRONG_SECRET, algorithm="HS256")
|
||||
with pytest.raises(ValueError, match="Token type must be 'refresh'"):
|
||||
service.verify_refresh_token(token)
|
||||
|
||||
def test_expired_refresh_token_raises(self):
|
||||
config = JWTConfig(secret_key=STRONG_SECRET, refresh_token_expire_days=0)
|
||||
def test_hs512_works(self):
|
||||
config = JWTConfig(secret_key=TEST_SECRET * 3, algorithm="HS512")
|
||||
service = JWTService(config)
|
||||
token = service.create_refresh_token(user_id="u1", session_id="s1")
|
||||
# 0天过期,应该立即使exp <= iat
|
||||
with pytest.raises(ExpiredSignatureError):
|
||||
service.verify_refresh_token(token)
|
||||
|
||||
|
||||
# ── JWTHandler 委托层测试 ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestJWTHandler:
|
||||
"""JWTHandler 委托层测试"""
|
||||
|
||||
def test_init_creates_handler(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET)
|
||||
assert handler is not None
|
||||
|
||||
def test_create_and_verify_access_token(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET)
|
||||
token = handler.create_access_token(user_id="u1", role="user")
|
||||
payload = handler.verify_access_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
assert payload["role"] == "user"
|
||||
|
||||
def test_verify_token_generic(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET)
|
||||
token = handler.create_access_token(user_id="u1")
|
||||
payload = handler.verify_token(token)
|
||||
token = service.create_access_token(user_id="u1")
|
||||
payload = service.verify_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
|
||||
def test_custom_algorithm(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET, algorithm="HS384")
|
||||
token = handler.create_access_token(user_id="u1")
|
||||
payload = handler.verify_access_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
def test_algorithm_mismatch_fails(self):
|
||||
config_hs256 = JWTConfig(secret_key=TEST_SECRET, algorithm="HS256")
|
||||
config_hs384 = JWTConfig(secret_key=TEST_SECRET, algorithm="HS384")
|
||||
service_256 = JWTService(config_hs256)
|
||||
service_384 = JWTService(config_hs384)
|
||||
|
||||
def test_custom_expire_minutes(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET, access_token_expire_minutes=45)
|
||||
token = handler.create_access_token(user_id="u1")
|
||||
payload = handler.verify_access_token(token)
|
||||
delta = payload["exp"] - payload["iat"]
|
||||
assert delta == 45 * 60
|
||||
|
||||
def test_additional_claims_passthrough(self):
|
||||
handler = JWTHandler(secret_key=STRONG_SECRET)
|
||||
extra = {"org_id": "org-1", "plan": "pro"}
|
||||
token = handler.create_access_token("u1", additional_claims={"org_id": "org-1"})
|
||||
payload = handler.verify_access_token(
|
||||
token := handler.create_access_token("u1", additional_claims={"org_id": "org-1"})
|
||||
)
|
||||
# 这里直接测试更简洁
|
||||
payload = handler.verify_access_token(handler.create_access_token("u1", additional_claims={"x": 1}))
|
||||
assert payload["x"] == 1
|
||||
token = service_256.create_access_token(user_id="u1")
|
||||
with pytest.raises(InvalidTokenError):
|
||||
service_384.verify_token(token)
|
||||
|
||||
|
||||
# ── 全局 JWT handler 测试 ───────────────────────────────────────────────────
|
||||
# ── 边界:空用户ID等 ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestGlobalJWTHandler:
|
||||
"""全局 JWT Handler 配置与获取测试"""
|
||||
class TestEdgeCases:
|
||||
def setup_method(self):
|
||||
self.service = JWTService(JWTConfig(secret_key=TEST_SECRET))
|
||||
|
||||
def test_configure_creates_handler(self):
|
||||
handler = configure_jwt_handler(secret_key=STRONG_SECRET)
|
||||
assert isinstance(handler, JWTHandler)
|
||||
def test_empty_user_id(self):
|
||||
token = self.service.create_access_token(user_id="")
|
||||
payload = self.service.verify_access_token(token)
|
||||
assert payload["sub"] == ""
|
||||
|
||||
def test_get_after_configure_works(self):
|
||||
configure_jwt_handler(secret_key=STRONG_SECRET)
|
||||
handler = get_jwt_handler()
|
||||
assert isinstance(handler, JWTHandler)
|
||||
token = handler.create_access_token(user_id="u1")
|
||||
payload = handler.verify_access_token(token)
|
||||
assert payload["sub"] == "u1"
|
||||
def test_long_user_id(self):
|
||||
long_id = "x" * 1000
|
||||
token = self.service.create_access_token(user_id=long_id)
|
||||
payload = self.service.verify_access_token(token)
|
||||
assert payload["sub"] == long_id
|
||||
|
||||
def test_get_before_configure_raises(self):
|
||||
# 重置全局状态(通过设置 None 模拟未配置)
|
||||
import packages.application.auth.jwt_handler as mod
|
||||
def test_special_chars_in_user_id(self):
|
||||
uid = "user@#$%^&*()_+-=[]{}|;:',.<>?/`~"
|
||||
token = self.service.create_access_token(user_id=uid)
|
||||
payload = self.service.verify_access_token(token)
|
||||
assert payload["sub"] == uid
|
||||
|
||||
mod._default_handler = None
|
||||
with pytest.raises(RuntimeError, match="JWT handler not configured"):
|
||||
get_jwt_handler()
|
||||
def test_unicode_user_id(self):
|
||||
uid = "用户_测试_123_🎉"
|
||||
token = self.service.create_access_token(user_id=uid)
|
||||
payload = self.service.verify_access_token(token)
|
||||
assert payload["sub"] == uid
|
||||
|
||||
def test_configure_returns_same_as_get(self):
|
||||
h1 = configure_jwt_handler(secret_key=STRONG_SECRET)
|
||||
h2 = get_jwt_handler()
|
||||
assert h1 is h2
|
||||
|
||||
def test_reconfigure_replaces_handler(self):
|
||||
h1 = configure_jwt_handler(secret_key=STRONG_SECRET)
|
||||
h2 = configure_jwt_handler(secret_key=STRONG_SECRET + "_new")
|
||||
assert h1 is not h2
|
||||
assert get_jwt_handler() is h2
|
||||
def test_many_additional_claims(self):
|
||||
claims = {f"key_{i}": f"value_{i}" for i in range(50)}
|
||||
token = self.service.create_access_token(user_id="u1", additional_claims=claims)
|
||||
payload = self.service.verify_access_token(token)
|
||||
for i in range(50):
|
||||
assert payload[f"key_{i}"] == f"value_{i}"
|
||||
|
||||
Reference in New Issue
Block a user