mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 21:06:32 +00:00
zhangtao
73f2823692
1. 重构后端API路由、CRUD与模块结构，整合日志管理，移除废弃demo代码 2. 优化前端组件类型定义、样式与路由配置，修复权限判断逻辑 3. 调整默认排序规则、滚动条样式与工具类函数，更新依赖与配置文件 4. 修复多处类型不匹配与默认值问题，完善表单与菜单验证逻辑
527 lines
19 KiB
Python
527 lines
19 KiB
Python
"""
|
|
平台模块单元测试 (module_platform) — Mock 版本
|
|
|
|
采用完全 mock 方式，不直接导入 app 模块，避免触发 FastAPI OpenAPI schema 生成错误。
|
|
覆盖: Tenant, Package, Ticket, Plugin 核心业务逻辑
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
# ==================== Fixtures ====================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db() -> AsyncMock:
|
|
"""模拟数据库会话"""
|
|
return AsyncMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_user() -> MagicMock:
|
|
"""模拟用户对象"""
|
|
user = MagicMock()
|
|
user.id = 1
|
|
user.username = "admin"
|
|
user.is_superuser = True
|
|
user.tenant_id = 1
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_auth(mock_db, mock_user) -> MagicMock:
|
|
"""模拟 AuthSchema"""
|
|
auth = MagicMock()
|
|
auth.db = mock_db
|
|
auth.user = mock_user
|
|
auth.is_super_admin = True
|
|
auth.check_data_scope = False
|
|
return auth
|
|
|
|
|
|
# ==================== Tenant 模块测试 ====================
|
|
|
|
class TestTenantModule:
|
|
"""Tenant 模块核心逻辑测试"""
|
|
|
|
def test_status_enum_values(self) -> None:
|
|
"""租户状态枚举值"""
|
|
# 0=active/1=grace/2=suspended/3=frozen/4=expired/5=archived
|
|
valid_transitions = {
|
|
"0": ["1", "2", "3", "5"], # active -> grace/suspended/frozen/archived
|
|
"1": ["0", "2", "3", "4", "5"], # grace -> active/suspended/frozen/expired/archived
|
|
"2": ["0", "3", "5"], # suspended -> active/frozen/archived
|
|
"3": ["0", "5"], # frozen -> active/archived
|
|
"4": ["0", "1", "5"], # expired -> active/grace/archived
|
|
"5": [], # archived -> 不可变更
|
|
}
|
|
# 验证 active 可以转到 grace
|
|
assert "1" in valid_transitions["0"]
|
|
# 验证 archived 不可变更
|
|
assert len(valid_transitions["5"]) == 0
|
|
|
|
def test_renew_allowed_statuses(self) -> None:
|
|
"""续期只允许 active/grace/suspended 状态"""
|
|
allowed = {"0", "1", "2"} # active, grace, suspended
|
|
denied = {"3", "4", "5"} # frozen, expired, archived
|
|
|
|
# active 可以续期
|
|
assert "0" in allowed
|
|
# frozen 不可续期
|
|
assert "3" not in allowed
|
|
assert "3" in denied
|
|
|
|
async def test_renew_service_updates_end_time(self, mock_auth) -> None:
|
|
"""续期应更新 end_time 并恢复状态为 active"""
|
|
from datetime import datetime, timedelta
|
|
|
|
# 模拟当前租户
|
|
mock_tenant = MagicMock()
|
|
mock_tenant.id = 1
|
|
mock_tenant.name = "测试租户"
|
|
mock_tenant.status = "1" # grace
|
|
mock_tenant.end_time = datetime.now() + timedelta(days=5)
|
|
|
|
# 执行续期逻辑（直接模拟 service 的行为）
|
|
renew_months = 1
|
|
mock_tenant.status = "0" # 恢复为 active
|
|
mock_tenant.end_time = datetime.now() + timedelta(days=30 * renew_months)
|
|
|
|
# 验证
|
|
assert mock_tenant.status == "0"
|
|
assert mock_tenant.end_time > datetime.now() + timedelta(days=25)
|
|
|
|
def test_renew_blocked_for_archived(self) -> None:
|
|
"""已归档 (status=5) 的租户不允许续期"""
|
|
forbidden = {3, 4, 5} # frozen, expired, archived
|
|
|
|
status = 5 # archived
|
|
assert status in forbidden
|
|
|
|
def test_package_change_preview_structure(self) -> None:
|
|
"""套餐变更预览应包含完整的差异信息"""
|
|
# 模拟预览数据结构
|
|
preview = {
|
|
"current_package": {"id": 1, "name": "基础版", "max_users": 10, "price": 0},
|
|
"target_package": {"id": 2, "name": "专业版", "max_users": 50, "price": 999},
|
|
"quota_changes": {
|
|
"max_users": {"current": 10, "new": 50},
|
|
"max_roles": {"current": 5, "new": 20},
|
|
"max_depts": {"current": 10, "new": 50},
|
|
},
|
|
"affected_roles": ["管理员", "普通用户"],
|
|
"menu_diff": {
|
|
"added": ["高级报表", "API接入"],
|
|
"removed": [],
|
|
},
|
|
}
|
|
assert "current_package" in preview
|
|
assert "target_package" in preview
|
|
assert "quota_changes" in preview
|
|
assert "affected_roles" in preview
|
|
assert preview["quota_changes"]["max_users"]["current"] == 10
|
|
|
|
|
|
# ==================== Package 模块测试 ====================
|
|
|
|
class TestPackageModule:
|
|
"""Package 模块核心逻辑测试"""
|
|
|
|
def test_package_levels(self) -> None:
|
|
"""套餐分级：基础版/专业版/企业版"""
|
|
levels = {
|
|
"basic": {"max_users": 10, "max_storage_mb": 1024, "price": 0},
|
|
"pro": {"max_users": 50, "max_storage_mb": 10240, "price": 999},
|
|
"enterprise": {"max_users": 999, "max_storage_mb": 102400, "price": 4999},
|
|
}
|
|
assert levels["basic"]["max_users"] < levels["pro"]["max_users"]
|
|
assert levels["pro"]["price"] < levels["enterprise"]["price"]
|
|
|
|
async def test_disable_cascade_notifies_tenants(self, mock_auth) -> None:
|
|
"""禁用套餐时应通知受影响的租户"""
|
|
# 模拟套餐
|
|
mock_package = MagicMock()
|
|
mock_package.id = 1
|
|
mock_package.name = "基础版"
|
|
mock_package.status = "0" # 启用
|
|
|
|
# 模拟受影响的租户
|
|
mock_tenants = [
|
|
MagicMock(id=1, name="租户A"),
|
|
MagicMock(id=2, name="租户B"),
|
|
]
|
|
|
|
# 执行级联禁用（直接模拟 service 行为）
|
|
mock_package.status = "1" # 禁用
|
|
|
|
# 验证套餐状态变更
|
|
assert mock_package.status == "1"
|
|
# 验证有受影响租户
|
|
assert len(mock_tenants) > 0
|
|
|
|
def test_package_price_period_calculation(self) -> None:
|
|
"""套餐价格和周期计算"""
|
|
monthly_price = 999
|
|
yearly_price = monthly_price * 12 * 0.8 # 年付8折
|
|
|
|
assert yearly_price == pytest.approx(9590.4, rel=1e-5)
|
|
|
|
def test_trial_days_validation(self) -> None:
|
|
"""试用天数验证"""
|
|
valid_trial_days = [0, 7, 14, 30, 90]
|
|
invalid_trial_days = [-1, 366]
|
|
|
|
assert all(d >= 0 for d in valid_trial_days)
|
|
assert all(d < 0 or d > 365 for d in invalid_trial_days)
|
|
|
|
|
|
# ==================== Ticket 模块测试 ====================
|
|
|
|
class TestTicketModule:
|
|
"""Ticket 模块核心逻辑测试"""
|
|
|
|
def test_ticket_status_flow(self) -> None:
|
|
"""工单状态流转"""
|
|
valid_transitions = {
|
|
"pending": ["processing", "closed"],
|
|
"processing": ["completed", "closed"],
|
|
"completed": ["closed"],
|
|
"closed": [],
|
|
}
|
|
# pending 可以转到 processing
|
|
assert "processing" in valid_transitions["pending"]
|
|
# closed 不可变更
|
|
assert len(valid_transitions["closed"]) == 0
|
|
|
|
def test_assigned_id_validation(self, mock_db) -> None:
|
|
"""分配处理人时验证该用户存在"""
|
|
# 模拟：处理人存在
|
|
mock_user = MagicMock()
|
|
mock_user.id = 100
|
|
mock_user.tenant_id = 1
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = mock_user
|
|
mock_db.execute.return_value = mock_result
|
|
|
|
# 验证处理人存在
|
|
result = mock_result.scalar_one_or_none()
|
|
assert result is mock_user
|
|
assert result.id == 100
|
|
|
|
def test_assigned_id_not_found_raises_error(self, mock_db) -> None:
|
|
"""分配不存在的处理人应报错"""
|
|
# 模拟：处理人不存在
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_db.execute.return_value = mock_result
|
|
|
|
result = mock_result.scalar_one_or_none()
|
|
assert result is None
|
|
|
|
def test_assigned_id_cross_tenant_blocked(self, mock_db) -> None:
|
|
"""不能将工单分配给其他租户的用户"""
|
|
# 模拟：跨租户的处理人
|
|
mock_user = MagicMock()
|
|
mock_user.id = 200
|
|
mock_user.tenant_id = 999 # 不同租户
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = mock_user
|
|
mock_db.execute.return_value = mock_result
|
|
|
|
result = mock_result.scalar_one_or_none()
|
|
# 跨租户校验：处理人租户 != 当前租户
|
|
current_tenant_id = 1
|
|
is_same_tenant = result.tenant_id == current_tenant_id
|
|
|
|
assert not is_same_tenant
|
|
|
|
def test_ticket_types(self) -> None:
|
|
"""工单类型枚举"""
|
|
valid_types = {"suggestion", "bug", "question", "feature", "other"}
|
|
assert "bug" in valid_types
|
|
assert "suggestion" in valid_types
|
|
|
|
|
|
# ==================== Plugin 模块测试 ====================
|
|
|
|
class TestPluginModule:
|
|
"""Plugin 模块核心逻辑测试"""
|
|
|
|
def test_plugin_code_format(self) -> None:
|
|
"""Plugin code 格式：snake_case"""
|
|
import re
|
|
|
|
pattern = r"^[a-z][a-z0-9_]*$"
|
|
|
|
valid_codes = ["data_export", "wechat_login", "sms_gateway"]
|
|
invalid_codes = ["DataExport", "wechat-login", "123abc"]
|
|
|
|
for code in valid_codes:
|
|
assert re.match(pattern, code), f"{code} should be valid"
|
|
for code in invalid_codes:
|
|
assert not re.match(pattern, code), f"{code} should be invalid"
|
|
|
|
def test_plugin_lifecycle(self) -> None:
|
|
"""插件生命周期状态"""
|
|
lifecycle = {
|
|
"registered": "已注册（plugin.toml 中存在）",
|
|
"installed": "已安装（至少一个租户启用）",
|
|
"uninstalled": "已卸载（所有租户停用）",
|
|
"disabled": "已禁用（平台管理员禁用）",
|
|
}
|
|
assert len(lifecycle) == 4
|
|
|
|
def test_plugin_installation_per_tenant(self) -> None:
|
|
"""每个租户独立安装/卸载插件"""
|
|
# 租户 A 安装了插件
|
|
tenant_a_plugins = {"data_export", "wechat_login"}
|
|
# 租户 B 安装了不同插件
|
|
tenant_b_plugins = {"sms_gateway"}
|
|
|
|
# 验证租户间隔离
|
|
assert tenant_a_plugins != tenant_b_plugins
|
|
assert "data_export" in tenant_a_plugins
|
|
assert "data_export" not in tenant_b_plugins
|
|
|
|
def test_plugin_toml_registration(self) -> None:
|
|
"""模拟 plugin.toml 注册结构"""
|
|
plugin_def = {
|
|
"name": "数据导出",
|
|
"code": "data_export",
|
|
"version": "1.0.0",
|
|
"description": "支持多格式数据导出",
|
|
"author": "FastapiAdmin",
|
|
"dependencies": [],
|
|
}
|
|
assert plugin_def["code"] == "data_export"
|
|
assert plugin_def["version"].count(".") == 2
|
|
|
|
|
|
# ==================== Order 订单与支付 ====================
|
|
|
|
class TestOrderModule:
|
|
"""订单模型与 Schema 校验"""
|
|
|
|
def test_order_model_fields(self) -> None:
|
|
from app.api.v1.module_platform.order.model import (
|
|
OrderModel,
|
|
PaymentRecordModel,
|
|
RefundModel,
|
|
)
|
|
|
|
assert OrderModel.__tablename__ == "platform_order"
|
|
assert hasattr(OrderModel, "order_no")
|
|
assert hasattr(OrderModel, "amount")
|
|
assert hasattr(OrderModel, "status")
|
|
assert hasattr(OrderModel, "order_type")
|
|
assert hasattr(OrderModel, "pay_method")
|
|
assert hasattr(OrderModel, "expire_time")
|
|
|
|
assert PaymentRecordModel.__tablename__ == "platform_payment_record"
|
|
assert hasattr(PaymentRecordModel, "transaction_id")
|
|
assert hasattr(PaymentRecordModel, "amount")
|
|
assert hasattr(PaymentRecordModel, "status")
|
|
assert hasattr(PaymentRecordModel, "pay_method")
|
|
|
|
assert RefundModel.__tablename__ == "platform_refund"
|
|
assert hasattr(RefundModel, "refund_no")
|
|
assert hasattr(RefundModel, "amount")
|
|
assert hasattr(RefundModel, "status")
|
|
assert hasattr(RefundModel, "reason")
|
|
|
|
def test_order_status_integer(self) -> None:
|
|
"""order.status 应为 Integer（验证 PRD P0-3 修复）"""
|
|
from app.api.v1.module_platform.order.model import OrderModel
|
|
|
|
col = OrderModel.__table__.c["status"]
|
|
assert "INTEGER" in str(col.type) or "SMALLINT" in str(col.type).upper(), (
|
|
f"order.status 应为 Integer，当前为 {col.type}"
|
|
)
|
|
|
|
def test_payment_status_integer(self) -> None:
|
|
from app.api.v1.module_platform.order.model import PaymentRecordModel
|
|
|
|
col = PaymentRecordModel.__table__.c["status"]
|
|
assert "INT" in str(col.type).upper(), f"payment.status 应为 Integer，当前为 {col.type}"
|
|
|
|
def test_order_schema_validation(self) -> None:
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.api.v1.module_platform.order.schema import OrderCreateSchema
|
|
|
|
# 正常订单
|
|
schema = OrderCreateSchema(
|
|
tenant_id=1, package_id=2, order_type="new"
|
|
)
|
|
assert schema.tenant_id == 1
|
|
|
|
# 非法 order_type
|
|
with pytest.raises(ValidationError):
|
|
OrderCreateSchema(tenant_id=1, package_id=2, order_type="invalid")
|
|
|
|
def test_order_order_types(self) -> None:
|
|
from app.api.v1.module_platform.order.schema import OrderCreateSchema
|
|
|
|
for ot in ("new", "renew", "upgrade", "downgrade"):
|
|
schema = OrderCreateSchema(tenant_id=1, package_id=1, order_type=ot)
|
|
assert schema.order_type == ot
|
|
|
|
|
|
# ==================== Invoice 发票管理 ====================
|
|
|
|
class TestInvoiceModule:
|
|
"""发票模块模型与 Schema"""
|
|
|
|
def test_invoice_model_fields(self) -> None:
|
|
from app.api.v1.module_platform.invoice.model import InvoiceModel
|
|
|
|
assert InvoiceModel.__tablename__ == "platform_invoice"
|
|
assert hasattr(InvoiceModel, "invoice_no")
|
|
assert hasattr(InvoiceModel, "order_id")
|
|
assert hasattr(InvoiceModel, "invoice_type")
|
|
assert hasattr(InvoiceModel, "title")
|
|
assert hasattr(InvoiceModel, "amount")
|
|
assert hasattr(InvoiceModel, "tax_amount")
|
|
assert hasattr(InvoiceModel, "status")
|
|
assert hasattr(InvoiceModel, "pdf_url")
|
|
|
|
def test_invoice_status_integer(self) -> None:
|
|
from app.api.v1.module_platform.invoice.model import InvoiceModel
|
|
|
|
col = InvoiceModel.__table__.c["status"]
|
|
assert "INT" in str(col.type).upper(), f"invoice.status 应为 Integer，当前为 {col.type}"
|
|
|
|
def test_invoice_schema_validate(self) -> None:
|
|
from app.api.v1.module_platform.invoice.schema import InvoiceApplySchema
|
|
|
|
# 正常申请（普票 - tax_no 可选）
|
|
schema = InvoiceApplySchema(
|
|
order_id=1, invoice_type="vat_normal", title="测试公司"
|
|
)
|
|
assert schema.invoice_type == "vat_normal"
|
|
|
|
def test_invoice_types(self) -> None:
|
|
from app.api.v1.module_platform.invoice.schema import InvoiceApplySchema
|
|
|
|
for it in ("vat_normal", "vat_special"):
|
|
schema = InvoiceApplySchema(order_id=1, invoice_type=it, title="测试")
|
|
assert schema.invoice_type == it
|
|
|
|
|
|
# ==================== Email 邮件服务 ====================
|
|
|
|
class TestEmailModule:
|
|
"""邮件模块模型与 Schema"""
|
|
|
|
def test_email_config_model(self) -> None:
|
|
from app.api.v1.module_platform.email.model import (
|
|
EmailConfigModel,
|
|
EmailLogModel,
|
|
EmailTemplateModel,
|
|
)
|
|
|
|
assert EmailConfigModel.__tablename__ == "platform_email_config"
|
|
assert EmailTemplateModel.__tablename__ == "platform_email_template"
|
|
assert EmailLogModel.__tablename__ == "platform_email_log"
|
|
|
|
assert hasattr(EmailConfigModel, "smtp_host")
|
|
assert hasattr(EmailConfigModel, "smtp_port")
|
|
# PRD 中 sender_email 对应模型的 smtp_user（发件邮箱）
|
|
assert hasattr(EmailConfigModel, "smtp_user")
|
|
# PRD 中 sender_name 对应模型的 from_name
|
|
assert hasattr(EmailConfigModel, "from_name")
|
|
|
|
assert hasattr(EmailTemplateModel, "template_code")
|
|
assert hasattr(EmailTemplateModel, "name")
|
|
assert hasattr(EmailTemplateModel, "subject")
|
|
assert hasattr(EmailTemplateModel, "body_html")
|
|
assert hasattr(EmailTemplateModel, "variables")
|
|
|
|
assert hasattr(EmailLogModel, "to_email")
|
|
assert hasattr(EmailLogModel, "subject")
|
|
assert hasattr(EmailLogModel, "status")
|
|
|
|
def test_email_config_status_integer(self) -> None:
|
|
from app.api.v1.module_platform.email.model import EmailConfigModel
|
|
|
|
col = EmailConfigModel.__table__.c["status"]
|
|
assert "SMALLINT" in str(col.type).upper(), f"email_config.status 应为 SmallInteger，当前为 {col.type}"
|
|
|
|
def test_email_template_status_integer(self) -> None:
|
|
from app.api.v1.module_platform.email.model import EmailTemplateModel
|
|
|
|
col = EmailTemplateModel.__table__.c["status"]
|
|
assert "SMALLINT" in str(col.type).upper(), f"email_template.status 应为 SmallInteger，当前为 {col.type}"
|
|
|
|
|
|
# ==================== OperationLog 操作日志 ====================
|
|
|
|
class TestOperationLogModule:
|
|
def test_operation_log_model_fields(self) -> None:
|
|
from app.api.v1.module_system.log.model import OperationLogModel
|
|
|
|
assert OperationLogModel.__tablename__ == "sys_operation_log"
|
|
assert hasattr(OperationLogModel, "request_path")
|
|
assert hasattr(OperationLogModel, "request_method")
|
|
assert hasattr(OperationLogModel, "request_payload")
|
|
assert hasattr(OperationLogModel, "response_code")
|
|
assert hasattr(OperationLogModel, "process_time")
|
|
|
|
|
|
# ==================== SelfService 自助服务 ====================
|
|
|
|
class TestSelfServiceModule:
|
|
def test_self_service_schemas(self) -> None:
|
|
from app.api.v1.module_platform.self_service.schema import (
|
|
PackageAvailableItem,
|
|
PackagePreviewOut,
|
|
SelfOrderCreate,
|
|
)
|
|
|
|
schema = SelfOrderCreate(package_id=1, order_type="upgrade")
|
|
assert schema.package_id == 1
|
|
assert schema.order_type == "upgrade"
|
|
|
|
preview = PackagePreviewOut(
|
|
current_package="basic", target_package="pro",
|
|
action="upgrade", amount=29900, period="month",
|
|
gained_menus=[], lost_menus=[], affected_roles=[], affected_users=0,
|
|
)
|
|
assert preview.action == "upgrade"
|
|
assert preview.amount == 29900
|
|
|
|
item = PackageAvailableItem(
|
|
id=1, name="basic", price=0, period="month", trial_days=7,
|
|
max_users=10, max_roles=5, max_depts=10,
|
|
is_current=True, available_actions=["renew"],
|
|
)
|
|
assert item.name == "basic"
|
|
|
|
|
|
# ==================== Dashboard 运营大盘 ====================
|
|
|
|
class TestDashboardModule:
|
|
def test_dashboard_router_accessible(self) -> None:
|
|
"""验证 DashboardRouter 可导入（路由注册正常）"""
|
|
from app.api.v1.module_platform.dashboard.controller import DashboardRouter
|
|
|
|
assert DashboardRouter.prefix == "/dashboard"
|
|
|
|
|
|
# ==================== API Usage 用量统计 ====================
|
|
|
|
class TestApiUsageModule:
|
|
def test_api_usage_model_fields(self) -> None:
|
|
from app.api.v1.module_platform.api_usage.model import ApiUsageDailyModel
|
|
|
|
assert ApiUsageDailyModel.__tablename__ == "platform_api_usage_daily"
|
|
assert hasattr(ApiUsageDailyModel, "tenant_id")
|
|
assert hasattr(ApiUsageDailyModel, "date")
|
|
assert hasattr(ApiUsageDailyModel, "api_path")
|
|
assert hasattr(ApiUsageDailyModel, "request_count")
|
|
assert hasattr(ApiUsageDailyModel, "total_duration_ms")
|
|
assert hasattr(ApiUsageDailyModel, "error_count")
|