taiji-AI-PAD/BACKEND_VERIFICATION.md

后端实现验证清单

本文档提供了一个系统的验证清单，帮助您确认所有后端功能都已正确实现并可以正常工作。

准备工作

环境配置

• Python 3.11+ 已安装
• Docker 已安装
• kubectl 已安装
• Azure CLI 已安装
• 已配置Azure订阅

数据库初始化

# 进入mcp-server目录
cd services/mcp-server

# 安装依赖
pip install -r requirements.txt

# 初始化数据库（会自动创建表和初始数据）
python -c "from database import init_db; import asyncio; asyncio.run(init_db())"

功能验证

✅ 1. 数据库模型验证

检查所有表是否正确创建：

-- 连接到PostgreSQL
psql "postgresql://taiji:By%40123456.@taijipda.postgres.database.azure.com:5432/postgres?sslmode=require"

-- 检查表
\dt

-- 应该看到以下表：
-- users
-- channels
-- agents
-- model_providers
-- resource_allocations
-- billing_records
-- recharge_records
-- applications
-- workflows
-- api_keys
-- gateway_apis
-- data_templates
-- tools
-- sessions
-- executions
-- audit_logs

验证表结构：

-- 检查User表
\d users

-- 验证余额和授信字段存在
SELECT id, name, email, balance, credit_limit, subscription_tier FROM users LIMIT 1;

-- 检查Channel表
\d channels

-- 检查Agent表
\d agents

-- 检查计费记录表
\d billing_records

✅ 2. 认证系统验证

启动服务器：

cd services/mcp-server
python main.py

测试登录（用户）：

curl -X POST http://localhost:8000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "email": "admin@taiji-ai.com",
    "password": "admin123",
    "role": "user"
  }'

预期响应：

{
  "success": true,
  "data": {
    "token": "eyJ...",
    "refreshToken": "eyJ...",
    "user": {
      "id": "...",
      "name": "系统管理员",
      "email": "admin@taiji-ai.com",
      "role": "super_admin"
    }
  }
}

测试API密钥：

# 使用上面获取的token
TOKEN="your-token-here"

curl -X GET http://localhost:8000/api/auth/keys/info \
  -H "Authorization: Bearer $TOKEN"

✅ 3. 用户侧平台API验证

测试仪表板统计：

curl -X GET http://localhost:8000/api/user/dashboard/stats \
  -H "Authorization: Bearer $TOKEN"

预期响应：

{
  "success": true,
  "data": {
    "activeAgents": 0,
    "totalRequests": 0,
    "euBalance": 0,
    "systemHealth": 98.5
  }
}

测试创建工作流（验证3节点限制）：

# 测试4个节点（应该失败）
curl -X POST http://localhost:8000/api/user/workflows/create \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "测试工作流",
    "description": "测试节点限制",
    "gateway": "MCP",
    "nodes": [
      {"agentId": "1", "agentType": "platform", "agentName": "Agent1", "order": 1},
      {"agentId": "2", "agentType": "platform", "agentName": "Agent2", "order": 2},
      {"agentId": "3", "agentType": "platform", "agentName": "Agent3", "order": 3},
      {"agentId": "4", "agentType": "platform", "agentName": "Agent4", "order": 4}
    ]
  }'

预期响应：

{
  "success": false,
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "工作流最多支持3个Agent节点"
  }
}

测试余额查询：

curl -X GET http://localhost:8000/api/user/billing/balance \
  -H "Authorization: Bearer $TOKEN"

✅ 4. 渠道合作伙伴API验证

创建测试渠道：

# 使用管理员账号
curl -X POST http://localhost:8000/api/admin/channels/create \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "测试渠道",
    "email": "channel@test.com",
    "password": "test123",
    "commissionRate": 10
  }'

渠道登录：

curl -X POST http://localhost:8000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "email": "channel@test.com",
    "password": "test123",
    "role": "channel"
  }'

创建租户：

curl -X POST http://localhost:8000/api/channel/tenants/create \
  -H "Authorization: Bearer $CHANNEL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "测试租户",
    "email": "tenant@test.com",
    "password": "test123",
    "subscriptionTier": "pro"
  }'

为租户充值：

curl -X POST http://localhost:8000/api/channel/tenants/{tenant_id}/recharge \
  -H "Authorization: Bearer $CHANNEL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "amount": 100
  }'

设置授信额度：

curl -X PUT http://localhost:8000/api/channel/tenants/{tenant_id}/credit \
  -H "Authorization: Bearer $CHANNEL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "creditLimit": 500
  }'

✅ 5. 超级管理员API验证

获取平台统计：

curl -X GET http://localhost:8000/api/admin/dashboard/stats \
  -H "Authorization: Bearer $ADMIN_TOKEN"

查看所有渠道：

curl -X GET http://localhost:8000/api/admin/channels \
  -H "Authorization: Bearer $ADMIN_TOKEN"

分配渠道资源：

curl -X PUT http://localhost:8000/api/admin/channels/{channel_id}/resources \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "models": ["model-id-1"],
    "agents": [
      {"agentId": "agent-id-1", "quantity": 10}
    ],
    "customAgentResources": {
      "cpu": 4,
      "memory": 8
    },
    "channelCredit": 10000
  }'

查看申请列表：

curl -X GET http://localhost:8000/api/admin/channels/applications \
  -H "Authorization: Bearer $ADMIN_TOKEN"

审批申请：

curl -X PUT http://localhost:8000/api/admin/channels/applications/{app_id}/review \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "approved": true,
    "reason": "审批通过"
  }'

✅ 6. 供应商管理API验证

创建模型供应商：

curl -X POST http://localhost:8000/api/providers/models/create \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "OpenAI",
    "provider": "openai",
    "apiUrl": "https://api.openai.com/v1",
    "apiKey": "sk-xxxxx",
    "supportedModels": ["gpt-4", "gpt-3.5-turbo"],
    "rpm": 3500,
    "tpm": 90000
  }'

获取供应商列表：

curl -X GET http://localhost:8000/api/providers/models \
  -H "Authorization: Bearer $ADMIN_TOKEN"

✅ 7. 计费功能验证

创建测试计费记录：

# Python脚本测试
import asyncio
from database import AsyncSessionLocal
from app.billing import create_billing_record

async def test_billing():
    async with AsyncSessionLocal() as db:
        # 创建计费记录（60秒 = 6 EU = ¥0.06）
        record = await create_billing_record(
            tenant_id="tenant-uuid",
            agent_id="agent-uuid",
            agent_name="测试Agent",
            duration_seconds=60,
            db=db,
            channel_id="channel-uuid"
        )
        
        print(f"EU: {record.eu}")  # 应该是6
        print(f"成本: {record.cost}")  # 应该是0.06

asyncio.run(test_billing())

验证EU计算：

from app.billing import calculate_eu, calculate_cost

# 测试EU计算
assert calculate_eu(5) == 1   # 不足10秒按1 EU
assert calculate_eu(10) == 1  # 10秒 = 1 EU
assert calculate_eu(15) == 2  # 15秒 = 2 EU
assert calculate_eu(60) == 6  # 60秒 = 6 EU

# 测试成本计算
from decimal import Decimal
assert calculate_cost(1) == Decimal("0.01")
assert calculate_cost(100) == Decimal("1.00")

print("✅ EU计算验证通过")

验证余额扣除：

from app.billing import deduct_balance, add_balance
from decimal import Decimal

async def test_balance():
    async with AsyncSessionLocal() as db:
        # 先充值
        success, msg = await add_balance("user-id", Decimal("100"), db)
        print(f"充值: {success}, {msg}")
        
        # 扣费
        success, msg = await deduct_balance("user-id", Decimal("10"), db)
        print(f"扣费: {success}, {msg}")

asyncio.run(test_balance())

✅ 8. Azure AKS部署验证

检查K8s配置文件：

# 验证YAML语法
kubectl apply --dry-run=client -f services/mcp-server/k8s/deployment.yaml
kubectl apply --dry-run=client -f services/mcp-server/k8s/ingress.yaml

构建Docker镜像：

cd services/mcp-server
docker build -t mcp-server:test .

# 测试镜像
docker run -p 8000:8000 -e DATABASE_URL="sqlite+aiosqlite:///./test.db" mcp-server:test

测试健康检查：

curl http://localhost:8000/health

预期响应：

{
  "status": "healthy",
  "database": "connected",
  "timestamp": "2025-12-25T..."
}

API文档验证

访问Swagger文档：

http://localhost:8000/docs

检查是否包含所有端点：

• /api/auth/* (6个端点)
• /api/user/* (12个端点)
• /api/channel/* (9个端点)
• /api/admin/* (11个端点)
• /api/providers/* (6个端点)

性能测试

并发测试

# 安装ab (Apache Bench)
# Ubuntu: apt-get install apache2-utils
# macOS: 已预装

# 测试健康检查端点
ab -n 1000 -c 10 http://localhost:8000/health

# 测试认证端点
ab -n 100 -c 5 -p login.json -T application/json http://localhost:8000/api/auth/login

数据库连接池测试

import asyncio
from database import check_db_connection

async def test_connections():
    tasks = [check_db_connection() for _ in range(20)]
    results = await asyncio.gather(*tasks)
    print(f"成功连接: {sum(results)}/20")

asyncio.run(test_connections())

安全验证

认证测试

测试未授权访问：

# 应该返回401
curl http://localhost:8000/api/user/dashboard/stats

测试错误的token：

# 应该返回401
curl -H "Authorization: Bearer invalid-token" \
  http://localhost:8000/api/user/dashboard/stats

SQL注入测试

# 尝试SQL注入（应该被参数化查询阻止）
curl -X POST http://localhost:8000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "email": "admin@taiji-ai.com\" OR \"1\"=\"1",
    "password": "anything",
    "role": "user"
  }'

密码哈希验证

from app.auth import get_password_hash, verify_password

# 测试密码哈希
password = "test123"
hashed = get_password_hash(password)
print(f"哈希长度: {len(hashed)}")  # 应该是60（bcrypt）

# 验证密码
assert verify_password(password, hashed) == True
assert verify_password("wrong", hashed) == False

print("✅ 密码哈希验证通过")

日志验证

检查日志文件：

tail -f logs/mcp-server.log

验证日志格式（应该是JSON）：

{
  "timestamp": "2025-12-25T10:00:00Z",
  "level": "INFO",
  "message": "Application started",
  "service": "mcp-server"
}

监控指标验证

访问Prometheus metrics：

curl http://localhost:8000/metrics

应该看到：

• http_requests_total
• http_request_duration_seconds
• database_connections
• 等等

最终检查清单

代码质量

• 所有文件无linter错误
• 代码遵循PEP 8规范
• 类型注解完整
• 文档字符串完整

功能完整性

• 所有API端点已实现
• 所有业务规则已实现
• 错误处理完善
• 数据验证完整

数据库

• 所有表已创建
• 索引已添加
• 关联关系正确
• 初始数据已加载

安全性

• JWT认证工作正常
• API密钥认证工作正常
• 密码正确哈希
• 敏感数据已加密
• CORS配置正确

性能

• 数据库连接池配置
• 查询性能优化
• 缓存机制（如需要）
• 并发处理能力

部署

• Docker镜像构建成功
• K8s配置有效
• 健康检查工作正常
• 环境变量配置正确
• Secrets已配置

文档

• API文档完整
• 部署文档清晰
• README更新
• 变更日志记录

问题反馈

如果在验证过程中发现任何问题，请记录：

1. 问题描述:
2. 重现步骤:
3. 预期结果:
4. 实际结果:
5. 错误日志:

---

验证完成日期: _______________ 验证人: _______________ 状态: [ ] 通过 [ ] 未通过