taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity

更新渠道申请

Browse Source
This commit is contained in:
Ubuntu
2026-01-05 13:03:15 +00:00
parent 23c84ea23e
commit 3a8e12657b
8 changed files with 1014 additions and 656 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plans/mcp-server-design-issues.md
+299
View File
@@ -0,0 +1,299 @@
# MCP-Server 设计问题审查报告
## 概述
本报告检查了 MCP-Server 的设计代码，发现了多个设计错误和冲突问题。
---
## 一、严重问题（会导致运行时错误）
### 1.1 PlatformAgentQuota 模型字段缺失 ❌
**问题描述**：`user.py` 中使用了 `PlatformAgentQuota` 的多个字段，但这些字段在 `models.py` 中**不存在**。
**models.py 中的定义（第 1095-1131 行）**：
```python
class PlatformAgentQuota(BaseModel, Base):
__tablename__ = "platform_agent_quotas"
target_id = Column(GUID(), nullable=False)
target_type = Column(String(20), nullable=False) # channel, tenant
template_name = Column(String(100), nullable=False)
pod_quota = Column(Integer, nullable=False, default=0)
pod_used = Column(Integer, default=0)
allocated_by = Column(GUID(), ForeignKey("users.id"))
allocated_at = Column(DateTime, default=datetime.utcnow)
```
**user.py 中使用的缺失字段**：
| 缺失字段 | 使用位置 | 用途 |
|----------|----------|------|
| `is_active` | user.py:835, 881, 1011 | 过滤活跃配额 |
| `agent_type` | user.py:844, 880, 1010 | Agent 类型标识 |
| `cpu_per_pod` | user.py:849, 909, 935 | 每个 Pod 的 CPU 配置 |
| `memory_per_pod` | user.py:850, 911, 936 | 每个 Pod 的内存配置 |
**影响**：
- 用户调用 `/api/user/platform-agents/available` 会报错
- 用户调用 `/api/user/platform-agents/use` 会报错
- 用户停止平台 Agent 时会报错
**修复建议**：在 `models.py` 的 `PlatformAgentQuota` 类中添加缺失字段：
```python
class PlatformAgentQuota(BaseModel, Base):
__tablename__ = "platform_agent_quotas"
target_id = Column(GUID(), nullable=False)
target_type = Column(String(20), nullable=False) # channel, tenant
template_name = Column(String(100), nullable=False)
# 新增字段
agent_type = Column(String(100)) # Agent 类型名称（与 template_name 可能相同）
is_active = Column(Boolean, default=True) # 是否活跃
cpu_per_pod = Column(String(20), default="100m") # 每个 Pod 的 CPU
memory_per_pod = Column(String(20), default="128Mi") # 每个 Pod 的内存
# 原有字段
pod_quota = Column(Integer, nullable=False, default=0)
pod_used = Column(Integer, default=0)
allocated_by = Column(GUID(), ForeignKey("users.id"))
allocated_at = Column(DateTime, default=datetime.utcnow)
```
---
### 1.2 AgentBillingRecord 模型字段缺失 ❌
**问题描述**：`user.py` 中使用了 `AgentBillingRecord` 的多个字段，但这些字段在 `models.py` 中**不存在**。
**models.py 中的定义（第 1134-1172 行）**：
```python
class AgentBillingRecord(BaseModel, Base):
__tablename__ = "agent_billing_records"
user_id = Column(GUID(), ForeignKey("users.id"), nullable=False)
channel_id = Column(GUID(), ForeignKey("channels.id"))
agent_name = Column(String(100), nullable=False)
agent_type = Column(String(20), nullable=False) # platform, custom
template_name = Column(String(100), nullable=False)
duration_seconds = Column(Integer, nullable=False)
cpu_seconds = Column(sa.Float, default=0)
memory_gb_seconds = Column(sa.Float, default=0)
request_count = Column(Integer, default=0)
cost = Column(sa.Numeric(12, 4), nullable=False)
currency = Column(String(10), default="EU")
period_start = Column(DateTime, nullable=False)
period_end = Column(DateTime, nullable=False)
```
**user.py 中使用的缺失字段**：
| 缺失字段 | 使用位置 | 用途 |
|----------|----------|------|
| `is_platform_agent` | user.py:933, 1054, 1207, 1264, 1469, 1659, 1682 | 区分平台/自定义 Agent |
| `start_time` | user.py:934, 1071, 1209, 1285, 1488, 1684 | 开始时间 |
| `end_time` | user.py:992, 1054, 1265, 1469, 1685 | 结束时间 |
| `cpu_used` | user.py:935, 1211, 1291, 1378, 1421, 1487, 1687 | CPU 使用量 |
| `memory_used` | user.py:936, 1212, 1291, 1379, 1424, 1488, 1688 | 内存使用量 |
| `eu_consumed` | user.py:1002, 1287, 1687 | EU 消耗量 |
**影响**：
- 用户使用平台 Agent 时计费记录创建失败
- 用户创建自定义 Agent 时计费记录创建失败
- 计费历史查询失败
**修复建议**：在 `models.py` 的 `AgentBillingRecord` 类中添加缺失字段：
```python
class AgentBillingRecord(BaseModel, Base):
__tablename__ = "agent_billing_records"
user_id = Column(GUID(), ForeignKey("users.id"), nullable=False)
channel_id = Column(GUID(), ForeignKey("channels.id"))
agent_name = Column(String(100), nullable=False)
agent_type = Column(String(20), nullable=False) # 模板名称
template_name = Column(String(100), nullable=False)
# 新增字段
is_platform_agent = Column(Boolean, nullable=False) # 是否为平台 Agent
start_time = Column(DateTime) # 开始时间
end_time = Column(DateTime) # 结束时间（None 表示运行中）
cpu_used = Column(String(20)) # CPU 使用量（如 "100m"）
memory_used = Column(String(20)) # 内存使用量（如 "128Mi"）
eu_consumed = Column(sa.Float, default=0) # EU 消耗量
# 原有字段
duration_seconds = Column(Integer) # 改为可空，运行中时为 None
cpu_seconds = Column(sa.Float, default=0)
memory_gb_seconds = Column(sa.Float, default=0)
request_count = Column(Integer, default=0)
cost = Column(sa.Numeric(12, 4)) # 改为可空
currency = Column(String(10), default="EU")
period_start = Column(DateTime) # 改为可空
period_end = Column(DateTime) # 改为可空
```
---
## 二、设计冲突问题
### 2.1 平台 Agent 模板硬编码 ⚠️
**问题描述**：平台 Agent 模板在 `admin.py` 和 `channel.py` 中**重复硬编码**，而不是从 Agent Manager 获取。
**位置**：
- [`admin.py:2507-2547`](services/mcp-server/app/routes/admin.py:2507) - `PLATFORM_AGENT_TEMPLATES`
- [`channel.py:1737-1777`](services/mcp-server/app/routes/channel.py:1737) - `PLATFORM_AGENT_TEMPLATES`
**问题**：
1. 两处代码需要保持同步，容易出错
2. 新增模板需要修改代码并重新部署
3. 与 Agent Manager 的模板管理功能冲突
**修复建议**：
1. 删除硬编码的 `PLATFORM_AGENT_TEMPLATES`
2. 调用 `agent_manager_client.list_platform_templates()` 获取模板
3. 或者在数据库中存储模板配置
### 2.2 字段命名不一致 ⚠️
**问题描述**：同一概念在不同地方使用不同的字段名。
| 概念 | models.py | user.py | 说明 |
|------|-----------|---------|------|
| Agent 类型 | `agent_type` | `agent_type` / `template_name` | 有时混用 |
| 开始时间 | `period_start` | `start_time` | 不一致 |
| 结束时间 | `period_end` | `end_time` | 不一致 |
**修复建议**：统一字段命名，或者在模型中同时保留两个字段作为别名。
---
## 三、业务逻辑问题
### 3.1 平台 Agent 启动时机不符合需求 ⚠️
**当前实现**：
- `user.py` 中的 `/api/user/platform-agents/use` 接口是**用户主动调用**才启动 Pod
**需求**：
- 平台 Agent 应该在**渠道分配给用户时立即启动**，用户可以直接使用
**影响位置**：
- [`channel.py`](services/mcp-server/app/routes/channel.py) 中的租户资源分配接口
**修复建议**：
在渠道分配平台 Agent 配额给租户时，自动调用 Agent Manager 创建 Pod：
```python
# channel.py 中分配平台 Agent 给租户时
async def allocate_platform_agent_to_tenant(...):
# 1. 创建配额记录
tenant_quota = PlatformAgentQuota(...)
# 2. 立即调用 Agent Manager 启动 Pod
client = get_agent_manager_client()
result = await client.create_platform_agent(
name=f"{template_name}-{tenant_id[:8]}",
template=template_name,
user_id=tenant_id,
channel_id=channel_id,
config=AgentConfig(
cpu_request=cpu_per_pod,
memory_request=memory_per_pod,
...
)
)
# 3. 更新配额使用量
tenant_quota.pod_used = 1
tenant_quota.pod_name = result.name
```
### 3.2 自定义 Agent 配额检查逻辑正确 ✅
**当前实现**：
- 渠道分配配额给租户时只记录配额，不启动 Pod
- 用户创建自定义 Agent 时检查配额并启动 Pod
**符合需求**：自定义 Agent 需要用户填写配置后才启动。
---
## 四、数据库迁移需求
根据上述问题，需要创建数据库迁移脚本：
### 4.1 迁移脚本 007_fix_platform_agent_quota.sql
```sql
-- 为 platform_agent_quotas 表添加缺失字段
ALTER TABLE platform_agent_quotas
ADD COLUMN IF NOT EXISTS agent_type VARCHAR(100),
ADD COLUMN IF NOT EXISTS is_active BOOLEAN DEFAULT TRUE,
ADD COLUMN IF NOT EXISTS cpu_per_pod VARCHAR(20) DEFAULT '100m',
ADD COLUMN IF NOT EXISTS memory_per_pod VARCHAR(20) DEFAULT '128Mi';
-- 创建索引
CREATE INDEX IF NOT EXISTS idx_platform_agent_quota_active
ON platform_agent_quotas(is_active);
```
### 4.2 迁移脚本 008_fix_agent_billing_record.sql
```sql
-- 为 agent_billing_records 表添加缺失字段
ALTER TABLE agent_billing_records
ADD COLUMN IF NOT EXISTS is_platform_agent BOOLEAN,
ADD COLUMN IF NOT EXISTS start_time TIMESTAMP,
ADD COLUMN IF NOT EXISTS end_time TIMESTAMP,
ADD COLUMN IF NOT EXISTS cpu_used VARCHAR(20),
ADD COLUMN IF NOT EXISTS memory_used VARCHAR(20),
ADD COLUMN IF NOT EXISTS eu_consumed FLOAT DEFAULT 0;
-- 修改原有字段为可空
ALTER TABLE agent_billing_records
ALTER COLUMN duration_seconds DROP NOT NULL,
ALTER COLUMN cost DROP NOT NULL,
ALTER COLUMN period_start DROP NOT NULL,
ALTER COLUMN period_end DROP NOT NULL;
-- 创建索引
CREATE INDEX IF NOT EXISTS idx_agent_billing_is_platform
ON agent_billing_records(is_platform_agent);
CREATE INDEX IF NOT EXISTS idx_agent_billing_end_time
ON agent_billing_records(end_time);
```
---
## 五、问题汇总
| 问题类型 | 问题描述 | 严重程度 | 状态 |
|----------|----------|----------|------|
| 模型字段缺失 | PlatformAgentQuota 缺少 is_active, agent_type, cpu_per_pod, memory_per_pod | 严重 | 待修复 |
| 模型字段缺失 | AgentBillingRecord 缺少 is_platform_agent, start_time, end_time, cpu_used, memory_used, eu_consumed | 严重 | 待修复 |
| 代码重复 | PLATFORM_AGENT_TEMPLATES 在两处硬编码 | 中等 | 待修复 |
| 字段命名不一致 | period_start/start_time, period_end/end_time | 低 | 待修复 |
| 业务逻辑 | 平台 Agent 启动时机不符合需求 | 中等 | 待修复 |
---
## 六、修复优先级
1. **高优先级**：修复 `PlatformAgentQuota` 和 `AgentBillingRecord` 模型字段缺失
2. **中优先级**：修改平台 Agent 启动时机（渠道分配时启动）
3. **低优先级**：移除硬编码模板，改为从 Agent Manager 获取
---
## 七、版本历史
| 版本 | 日期 | 说明 |
|------|------|------|
| v1.0 | 2026-01-05 | 初始审查报告 |
services/mcp-server/app/routes/admin.py
+44 -4
View File
@@ -677,10 +677,50 @@ async def list_channels(
for alloc in allocations:
# 查询Agent的资源配置
agent_result = await db.execute(
select(Agent).where(Agent.id == alloc.resource_id)
)
agent = agent_result.scalar_one_or_none()
# resource_id 可能是 UUID 字符串或模板名称（如 'code-reviewer'）
agent = None
# 首先尝试按 UUID 查询
try:
resource_uuid = uuid.UUID(alloc.resource_id)
agent_result = await db.execute(
select(Agent).where(Agent.id == resource_uuid)
)
agent = agent_result.scalar_one_or_none()
except (ValueError, TypeError):
# 如果不是有效的 UUID，按名称查询
pass
# 如果按 UUID 没找到，尝试按名称查询
if not agent:
agent_result = await db.execute(
select(Agent).where(Agent.name == alloc.resource_id)
)
agent = agent_result.scalar_one_or_none()
# 如果仍然没找到，检查是否是平台 Agent 模板
if not agent and alloc.resource_id in PLATFORM_AGENT_TEMPLATES:
template = PLATFORM_AGENT_TEMPLATES[alloc.resource_id]
# 使用模板的默认资源配置
quantity = alloc.quantity or 1
# 解析 CPU (如 "500m" -> 0.5 核)
cpu_limit = template.get("cpuLimit", "0")
if isinstance(cpu_limit, str) and cpu_limit.endswith("m"):
total_cpu += float(cpu_limit[:-1]) / 1000 * quantity
elif cpu_limit:
try:
total_cpu += float(cpu_limit) * quantity
except ValueError:
pass
# 解析内存 (如 "512Mi" -> 0.5 GB)
memory_limit = template.get("memoryLimit", "0")
if isinstance(memory_limit, str):
if memory_limit.endswith("Mi"):
total_memory += float(memory_limit[:-2]) / 1024 * quantity
elif memory_limit.endswith("Gi"):
total_memory += float(memory_limit[:-2]) * quantity
continue
if agent:
quantity = alloc.quantity or 1
total_cpu += float(agent.cpu or 0) * quantity
services/mcp-server/migrations/007_change_resource_id_to_string.sql
+21
View File
@@ -0,0 +1,21 @@
-- Migration: 007_change_resource_id_to_string
-- Description: 将 resource_allocations 表的 resource_id 字段从 UUID 改为 VARCHAR(100)
-- 原因: resource_id 需要存储平台 Agent 模板名称（如 'code-reviewer'）而不仅仅是 UUID
-- Date: 2026-01-05
-- 1. 删除旧索引（如果存在）
DROP INDEX IF EXISTS idx_resource_allocation_resource;
-- 2. 修改 resource_id 字段类型
-- 注意：PostgreSQL 需要使用 USING 子句来转换类型
ALTER TABLE resource_allocations
ALTER COLUMN resource_id TYPE VARCHAR(100)
USING resource_id::VARCHAR(100);
-- 3. 重新创建索引
CREATE INDEX idx_resource_allocation_resource ON resource_allocations(resource_id, resource_type);
-- 验证修改
-- SELECT column_name, data_type, character_maximum_length
-- FROM information_schema.columns
-- WHERE table_name = 'resource_allocations' AND column_name = 'resource_id';
services/mcp-server/migrations/run_007_migration.py
+98
View File
@@ -0,0 +1,98 @@
#!/usr/bin/env python3
"""
执行数据库迁移：将 resource_allocations.resource_id 从 UUID 改为 VARCHAR(100)
"""
import asyncio
import os
import sys
import ssl
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
async def run_migration():
"""执行迁移"""
# 从环境变量获取数据库 URL
database_url = os.getenv("DATABASE_URL") or os.getenv("ASYNC_DATABASE_URL")
if not database_url:
print("错误: 请设置 DATABASE_URL 或 ASYNC_DATABASE_URL 环境变量")
sys.exit(1)
# 确保使用 asyncpg 驱动
if database_url.startswith("postgresql://") and "+asyncpg" not in database_url:
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
# 移除 URL 中的 sslmode 参数（asyncpg 使用不同的 SSL 配置方式）
if "sslmode=" in database_url:
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
parsed = urlparse(database_url)
query_params = parse_qs(parsed.query)
query_params.pop('sslmode', None)
new_query = urlencode(query_params, doseq=True)
database_url = urlunparse(parsed._replace(query=new_query))
print(f"连接数据库...")
# 创建 SSL 上下文
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
engine = create_async_engine(
database_url,
echo=True,
connect_args={"ssl": ssl_context}
)
try:
async with engine.begin() as conn:
print("执行迁移...")
# 1. 删除旧索引
print("步骤 1: 删除旧索引...")
await conn.execute(text("DROP INDEX IF EXISTS idx_resource_allocation_resource"))
# 2. 修改字段类型
print("步骤 2: 修改 resource_id 字段类型...")
await conn.execute(text("""
ALTER TABLE resource_allocations
ALTER COLUMN resource_id TYPE VARCHAR(100)
USING resource_id::VARCHAR(100)
"""))
# 3. 重新创建索引
print("步骤 3: 重新创建索引...")
await conn.execute(text("""
CREATE INDEX IF NOT EXISTS idx_resource_allocation_resource
ON resource_allocations(resource_id, resource_type)
"""))
print("迁移成功完成！")
# 验证修改
result = await conn.execute(text("""
SELECT column_name, data_type, character_maximum_length
FROM information_schema.columns
WHERE table_name = 'resource_allocations' AND column_name = 'resource_id'
"""))
row = result.fetchone()
if row:
print(f"验证结果: column={row[0]}, type={row[1]}, max_length={row[2]}")
except Exception as e:
print(f"迁移失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
finally:
await engine.dispose()
if __name__ == "__main__":
asyncio.run(run_migration())
services/mcp-server/models.py
+13 -5
View File
@@ -190,9 +190,10 @@ class Agent(BaseModel, Base):
owner = relationship("User", back_populates="agents")
executions = relationship("Execution", back_populates="agent", cascade="all, delete-orphan")
resource_allocations = relationship("ResourceAllocation",
foreign_keys="[ResourceAllocation.resource_id]",
primaryjoin="and_(Agent.id==ResourceAllocation.resource_id, ResourceAllocation.resource_type=='agent')")
# 注意：resource_allocations 关系已移除，因为 resource_id 现在是 String 类型
# 可以存储模板名称（如 'code-reviewer'）或 UUID 字符串
# 如需查询 Agent 的资源分配，请使用显式查询：
# select(ResourceAllocation).where(ResourceAllocation.resource_id == str(agent.id))
# 索引和约束
__table_args__ = (
@@ -340,6 +341,7 @@ class Channel(BaseModel, Base):
# 关联关系
tenants = relationship("User", back_populates="channel")
# 注意：resource_allocations 关系使用 target_id（GUID 类型）进行关联，仍然有效
resource_allocations = relationship("ResourceAllocation",
foreign_keys="[ResourceAllocation.target_id]",
primaryjoin="and_(Channel.id==ResourceAllocation.target_id, ResourceAllocation.target_type=='channel')")
@@ -374,14 +376,20 @@ class ModelProvider(BaseModel, Base):
class ResourceAllocation(BaseModel, Base):
"""资源分配表（渠道和租户的资源分配）"""
"""资源分配表（渠道和租户的资源分配）
注意：resource_id 字段使用 String 类型，因为它可能存储：
- 平台 Agent 模板名称（如 'code-reviewer', 'gpt-assistant'）
- 模型供应商 UUID
- 其他资源标识符
"""
__tablename__ = "resource_allocations"
target_id = Column(GUID(), nullable=False)
target_type = Column(String(20), nullable=False) # channel, tenant
resource_type = Column(String(20), nullable=False) # agent, model
resource_id = Column(GUID(), nullable=False)
resource_id = Column(String(100), nullable=False) # 资源标识符（模板名称或UUID字符串）
# Agent资源配置
quantity = Column(Integer) # Agent数量
services/mcp-server/tests/test_agent_management.py
-647
View File
@@ -1,647 +0,0 @@
"""
Agent 管理功能测试
测试内容：
1. 平台 Agent 管理
2. 自定义 Agent 管理
3. 资源申请审批
4. Agent 计费
"""
import pytest
import uuid
from datetime import datetime, timedelta
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock, patch
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy.ext.asyncio import AsyncSession
# ============= 测试数据 =============
def create_test_user(
user_id: str = None,
name: str = "测试用户",
email: str = "test@example.com",
role: str = "user",
channel_id: str = None,
):
"""创建测试用户对象"""
return MagicMock(
id=uuid.UUID(user_id) if user_id else uuid.uuid4(),
name=name,
email=email,
role=role,
channel_id=uuid.UUID(channel_id) if channel_id else uuid.uuid4(),
status="active",
balance=Decimal("100.00"),
credit_limit=Decimal("50.00"),
)
def create_test_channel(
channel_id: str = None,
name: str = "测试渠道",
):
"""创建测试渠道对象"""
return MagicMock(
id=uuid.UUID(channel_id) if channel_id else uuid.uuid4(),
name=name,
status="active",
)
def create_test_quota(
target_id: str,
target_type: str = "tenant",
template_name: str = "gpt-assistant",
pod_quota: int = 5,
pod_used: int = 0,
):
"""创建测试配额对象"""
return MagicMock(
id=uuid.uuid4(),
target_id=uuid.UUID(target_id),
target_type=target_type,
template_name=template_name,
pod_quota=pod_quota,
pod_used=pod_used,
allocated_at=datetime.utcnow(),
)
def create_test_billing_record(
user_id: str,
agent_type: str = "platform",
template_name: str = "gpt-assistant",
duration: int = 3600,
cost: float = 0.5,
):
"""创建测试计费记录"""
return MagicMock(
id=uuid.uuid4(),
tenant_id=uuid.UUID(user_id),
billing_type="agent",
agent_type=agent_type,
template_name=template_name,
agent_name=f"test-{template_name}",
instance_id=f"instance-{uuid.uuid4().hex[:8]}",
duration=duration,
eu=duration // 10,
cost=Decimal(str(cost)),
status="completed",
timestamp=datetime.utcnow(),
)
# ============= 计费逻辑测试 =============
class TestAgentBilling:
"""Agent 计费逻辑测试"""
def test_calculate_platform_agent_cost(self):
"""测试平台 Agent 计费计算"""
from app.billing import calculate_platform_agent_cost, PLATFORM_AGENT_PRICING
# 测试 gpt-assistant 模板，运行 1 小时
cost = calculate_platform_agent_cost("gpt-assistant", 3600)
expected_price = PLATFORM_AGENT_PRICING.get("gpt-assistant", 0.5)
assert cost == expected_price
# 测试 code-reviewer 模板，运行 2 小时
cost = calculate_platform_agent_cost("code-reviewer", 7200)
expected_price = PLATFORM_AGENT_PRICING.get("code-reviewer", 0.8) * 2
assert cost == expected_price
# 测试未知模板，使用默认价格
cost = calculate_platform_agent_cost("unknown-template", 3600)
assert cost == 0.5 # 默认价格
def test_calculate_custom_agent_cost(self):
"""测试自定义 Agent 计费计算"""
from app.billing import calculate_agent_cost_by_resources, AGENT_RESOURCE_PRICING
# 测试 1 核 CPU + 1GB 内存，运行 1 小时
cost = calculate_agent_cost_by_resources(
cpu_cores=1.0,
memory_gb=1.0,
duration_seconds=3600
)
cpu_price = AGENT_RESOURCE_PRICING["cpu_per_core_hour"]
memory_price = AGENT_RESOURCE_PRICING["memory_per_gb_hour"]
expected_cost = cpu_price + memory_price
assert abs(cost - expected_cost) < 0.001
# 测试 2 核 CPU + 4GB 内存，运行 30 分钟
cost = calculate_agent_cost_by_resources(
cpu_cores=2.0,
memory_gb=4.0,
duration_seconds=1800
)
expected_cost = (cpu_price * 2 + memory_price * 4) * 0.5
assert abs(cost - expected_cost) < 0.001
def test_parse_cpu_to_cores(self):
"""测试 CPU 解析"""
from app.billing import _parse_cpu_to_cores
# 测试毫核
assert _parse_cpu_to_cores("100m") == 0.1
assert _parse_cpu_to_cores("500m") == 0.5
assert _parse_cpu_to_cores("1000m") == 1.0
# 测试核心数
assert _parse_cpu_to_cores("1") == 1.0
assert _parse_cpu_to_cores("2") == 2.0
assert _parse_cpu_to_cores("0.5") == 0.5
# 测试数字输入
assert _parse_cpu_to_cores(1) == 1.0
assert _parse_cpu_to_cores(0.5) == 0.5
def test_parse_memory_to_gb(self):
"""测试内存解析"""
from app.billing import _parse_memory_to_gb
# 测试 Mi
assert _parse_memory_to_gb("128Mi") == 0.125
assert _parse_memory_to_gb("512Mi") == 0.5
assert _parse_memory_to_gb("1024Mi") == 1.0
# 测试 Gi
assert _parse_memory_to_gb("1Gi") == 1.0
assert _parse_memory_to_gb("2Gi") == 2.0
assert _parse_memory_to_gb("0.5Gi") == 0.5
# 测试数字输入（假设为 GB）
assert _parse_memory_to_gb(1) == 1.0
assert _parse_memory_to_gb(0.5) == 0.5
# ============= 配额管理测试 =============
class TestQuotaManagement:
"""配额管理测试"""
@pytest.mark.asyncio
async def test_check_platform_agent_quota(self):
"""测试平台 Agent 配额检查"""
# 模拟配额：5 个 Pod，已使用 3 个
quota = create_test_quota(
target_id=str(uuid.uuid4()),
template_name="gpt-assistant",
pod_quota=5,
pod_used=3,
)
# 剩余 2 个，请求 1 个应该成功
remaining = quota.pod_quota - quota.pod_used
assert remaining >= 1
# 请求 3 个应该失败
assert remaining < 3
@pytest.mark.asyncio
async def test_check_custom_agent_quota(self):
"""测试自定义 Agent 配额检查"""
# 模拟配额：4 核 CPU，8GB 内存
quota = MagicMock(
cpu_quota=Decimal("4.0"),
memory_quota=Decimal("8.0"),
cpu_used=Decimal("1.0"),
memory_used=Decimal("2.0"),
)
# 剩余 3 核 CPU，6GB 内存
cpu_remaining = float(quota.cpu_quota - quota.cpu_used)
memory_remaining = float(quota.memory_quota - quota.memory_used)
assert cpu_remaining == 3.0
assert memory_remaining == 6.0
# 请求 2 核 CPU，4GB 内存应该成功
assert cpu_remaining >= 2.0
assert memory_remaining >= 4.0
# 请求 4 核 CPU 应该失败
assert cpu_remaining < 4.0
# ============= 资源申请测试 =============
class TestResourceApplication:
"""资源申请测试"""
def test_create_platform_agent_application(self):
"""测试创建平台 Agent 申请"""
application = MagicMock(
id=uuid.uuid4(),
channel_id=uuid.uuid4(),
resource_type="platform_agent",
template_name="gpt-assistant",
requested_pod_quota=10,
reason="业务需要",
status="pending",
created_at=datetime.utcnow(),
)
assert application.resource_type == "platform_agent"
assert application.status == "pending"
assert application.requested_pod_quota == 10
def test_create_custom_agent_quota_application(self):
"""测试创建自定义 Agent 配额申请"""
application = MagicMock(
id=uuid.uuid4(),
channel_id=uuid.uuid4(),
resource_type="custom_agent_quota",
requested_cpu_quota=Decimal("8.0"),
requested_memory_quota=Decimal("16.0"),
reason="扩展业务",
status="pending",
created_at=datetime.utcnow(),
)
assert application.resource_type == "custom_agent_quota"
assert application.status == "pending"
assert float(application.requested_cpu_quota) == 8.0
assert float(application.requested_memory_quota) == 16.0
def test_approve_application(self):
"""测试审批申请"""
application = MagicMock(
id=uuid.uuid4(),
status="pending",
requested_pod_quota=10,
approved_pod_quota=None,
reviewed_by=None,
review_reason=None,
reviewed_at=None,
)
# 模拟审批
application.status = "approved"
application.approved_pod_quota = 8 # 批准 8 个，少于请求的 10 个
application.reviewed_by = str(uuid.uuid4())
application.review_reason = "批准 8 个 Pod"
application.reviewed_at = datetime.utcnow()
assert application.status == "approved"
assert application.approved_pod_quota == 8
assert application.reviewed_at is not None
def test_reject_application(self):
"""测试拒绝申请"""
application = MagicMock(
id=uuid.uuid4(),
status="pending",
)
# 模拟拒绝
application.status = "rejected"
application.review_reason = "资源不足"
application.reviewed_at = datetime.utcnow()
assert application.status == "rejected"
assert application.review_reason == "资源不足"
# ============= Agent Manager 客户端测试 =============
class TestAgentManagerClient:
"""Agent Manager 客户端测试"""
@pytest.mark.asyncio
async def test_list_platform_templates(self):
"""测试获取平台 Agent 模板列表"""
from app.agent_manager_client import AgentManagerClient
client = AgentManagerClient()
with patch.object(client, '_request', new_callable=AsyncMock) as mock_request:
mock_request.return_value = {
"templates": [
{
"name": "gpt-assistant",
"displayName": "GPT 智能助手",
"category": "assistant",
},
{
"name": "code-reviewer",
"displayName": "代码审查助手",
"category": "development",
},
]
}
result = await client.list_platform_templates()
assert "templates" in result
assert len(result["templates"]) == 2
assert result["templates"][0]["name"] == "gpt-assistant"
@pytest.mark.asyncio
async def test_create_platform_agent(self):
"""测试创建平台 Agent"""
from app.agent_manager_client import AgentManagerClient
client = AgentManagerClient()
with patch.object(client, '_request', new_callable=AsyncMock) as mock_request:
mock_request.return_value = {
"instanceName": "gpt-assistant-abc123",
"status": "running",
"endpoint": "http://gpt-assistant-abc123.agents.svc.cluster.local",
}
result = await client.create_platform_agent(
user_id="user-123",
template_name="gpt-assistant",
instance_name="gpt-assistant-abc123",
)
assert result["instanceName"] == "gpt-assistant-abc123"
assert result["status"] == "running"
@pytest.mark.asyncio
async def test_create_custom_agent(self):
"""测试创建自定义 Agent"""
from app.agent_manager_client import AgentManagerClient
client = AgentManagerClient()
with patch.object(client, '_request', new_callable=AsyncMock) as mock_request:
mock_request.return_value = {
"name": "my-custom-agent",
"status": "running",
"endpoint": "http://my-custom-agent.user-ns.svc.cluster.local",
}
result = await client.create_custom_agent(
user_id="user-123",
template_name="openai-compatible",
agent_name="my-custom-agent",
cpu_request="500m",
memory_request="512Mi",
env_vars={
"OPENAI_API_KEY": "sk-xxx",
"OPENAI_API_BASE": "https://api.openai.com/v1",
},
)
assert result["name"] == "my-custom-agent"
assert result["status"] == "running"
@pytest.mark.asyncio
async def test_scale_custom_agent(self):
"""测试扩缩容自定义 Agent"""
from app.agent_manager_client import AgentManagerClient
client = AgentManagerClient()
with patch.object(client, '_request', new_callable=AsyncMock) as mock_request:
mock_request.return_value = {
"name": "my-custom-agent",
"replicas": 3,
"status": "scaling",
}
result = await client.scale_custom_agent(
user_id="user-123",
agent_name="my-custom-agent",
replicas=3,
)
assert result["replicas"] == 3
assert result["status"] == "scaling"
# ============= API 路由测试 =============
class TestUserAgentAPIs:
"""用户 Agent API 测试"""
@pytest.mark.asyncio
async def test_list_available_platform_agents(self):
"""测试获取可用平台 Agent 列表"""
# 模拟用户有配额
quotas = [
create_test_quota(
target_id=str(uuid.uuid4()),
template_name="gpt-assistant",
pod_quota=5,
pod_used=2,
),
create_test_quota(
target_id=str(uuid.uuid4()),
template_name="code-reviewer",
pod_quota=3,
pod_used=0,
),
]
# 验证返回数据格式
result = []
for quota in quotas:
result.append({
"templateName": quota.template_name,
"podQuota": quota.pod_quota,
"podUsed": quota.pod_used,
"podRemaining": quota.pod_quota - quota.pod_used,
})
assert len(result) == 2
assert result[0]["podRemaining"] == 3
assert result[1]["podRemaining"] == 3
@pytest.mark.asyncio
async def test_use_platform_agent(self):
"""测试使用平台 Agent"""
# 模拟请求
request = {
"templateName": "gpt-assistant",
"agentType": "chat",
}
# 验证请求格式
assert "templateName" in request
assert request["templateName"] == "gpt-assistant"
class TestChannelAgentAPIs:
"""渠道 Agent API 测试"""
@pytest.mark.asyncio
async def test_allocate_platform_agent_to_tenant(self):
"""测试分配平台 Agent 给租户"""
# 模拟渠道配额
channel_quota = create_test_quota(
target_id=str(uuid.uuid4()),
target_type="channel",
template_name="gpt-assistant",
pod_quota=20,
pod_used=5,
)
# 模拟分配请求
request = {
"templateName": "gpt-assistant",
"podQuota": 5,
}
# 验证配额足够
remaining = channel_quota.pod_quota - channel_quota.pod_used
assert remaining >= request["podQuota"]
@pytest.mark.asyncio
async def test_get_agent_billing_stats(self):
"""测试获取 Agent 计费统计"""
# 模拟计费记录
records = [
create_test_billing_record(
user_id=str(uuid.uuid4()),
agent_type="platform",
template_name="gpt-assistant",
duration=3600,
cost=0.5,
),
create_test_billing_record(
user_id=str(uuid.uuid4()),
agent_type="custom",
template_name="openai-compatible",
duration=7200,
cost=1.2,
),
]
# 计算统计
total_cost = sum(float(r.cost) for r in records)
total_duration = sum(r.duration for r in records)
assert total_cost == 1.7
assert total_duration == 10800
# ============= 集成测试 =============
class TestIntegration:
"""集成测试"""
@pytest.mark.asyncio
async def test_full_platform_agent_workflow(self):
"""测试完整的平台 Agent 工作流程"""
# 1. 渠道申请平台 Agent
application = MagicMock(
id=uuid.uuid4(),
channel_id=uuid.uuid4(),
resource_type="platform_agent",
template_name="gpt-assistant",
requested_pod_quota=10,
status="pending",
)
# 2. 管理员审批
application.status = "approved"
application.approved_pod_quota = 10
# 3. 渠道获得配额
channel_quota = create_test_quota(
target_id=str(application.channel_id),
target_type="channel",
template_name="gpt-assistant",
pod_quota=10,
pod_used=0,
)
# 4. 渠道分配给租户
tenant_id = str(uuid.uuid4())
tenant_quota = create_test_quota(
target_id=tenant_id,
target_type="tenant",
template_name="gpt-assistant",
pod_quota=5,
pod_used=0,
)
# 5. 租户使用平台 Agent
tenant_quota.pod_used = 1
# 6. 创建计费记录
billing_record = create_test_billing_record(
user_id=tenant_id,
agent_type="platform",
template_name="gpt-assistant",
)
# 验证流程
assert application.status == "approved"
assert channel_quota.pod_quota == 10
assert tenant_quota.pod_used == 1
assert billing_record.agent_type == "platform"
@pytest.mark.asyncio
async def test_full_custom_agent_workflow(self):
"""测试完整的自定义 Agent 工作流程"""
# 1. 渠道申请自定义 Agent 配额
application = MagicMock(
id=uuid.uuid4(),
channel_id=uuid.uuid4(),
resource_type="custom_agent_quota",
requested_cpu_quota=Decimal("8.0"),
requested_memory_quota=Decimal("16.0"),
status="pending",
)
# 2. 管理员审批
application.status = "approved"
application.approved_cpu_quota = Decimal("8.0")
application.approved_memory_quota = Decimal("16.0")
# 3. 渠道获得配额
channel_quota = MagicMock(
channel_id=application.channel_id,
cpu_quota=Decimal("8.0"),
memory_quota=Decimal("16.0"),
cpu_allocated=Decimal("0"),
memory_allocated=Decimal("0"),
)
# 4. 渠道分配给租户
tenant_id = str(uuid.uuid4())
tenant_quota = MagicMock(
tenant_id=uuid.UUID(tenant_id),
cpu_quota=Decimal("4.0"),
memory_quota=Decimal("8.0"),
cpu_used=Decimal("0"),
memory_used=Decimal("0"),
)
# 5. 租户创建自定义 Agent
tenant_quota.cpu_used = Decimal("1.0")
tenant_quota.memory_used = Decimal("2.0")
# 6. 创建计费记录
billing_record = create_test_billing_record(
user_id=tenant_id,
agent_type="custom",
template_name="openai-compatible",
)
# 验证流程
assert application.status == "approved"
assert float(channel_quota.cpu_quota) == 8.0
assert float(tenant_quota.cpu_used) == 1.0
assert billing_record.agent_type == "custom"
# ============= 运行测试 =============
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])
test_allocate_platform_agent.py
+92
View File
@@ -0,0 +1,92 @@
#!/usr/bin/env python3
"""测试分配平台 Agent 给租户的接口"""
import requests
import json
BASE_URL = "http://localhost:8002"
def test_allocate_platform_agent():
# 1. 渠道管理员登录
print("=" * 50)
print("1. 渠道管理员登录")
login_resp = requests.post(
f"{BASE_URL}/api/channel/auth/login",
json={"email": "66@66.com", "password": "66"}
)
print(f"状态码: {login_resp.status_code}")
print(f"响应: {json.dumps(login_resp.json(), indent=2, ensure_ascii=False)}")
if login_resp.status_code != 200:
print("登录失败!")
return
token = login_resp.json().get("token")
headers = {"Authorization": f"Bearer {token}"}
# 解析 JWT token 查看内容
import base64
parts = token.split(".")
if len(parts) >= 2:
payload = parts[1]
# 添加填充
payload += "=" * (4 - len(payload) % 4)
decoded = base64.urlsafe_b64decode(payload)
print(f"\nJWT Payload: {decoded.decode('utf-8')}")
# 2. 获取渠道下的租户列表
print("\n" + "=" * 50)
print("2. 获取渠道下的租户列表")
tenants_resp = requests.get(
f"{BASE_URL}/api/channel/tenants",
headers=headers
)
print(f"状态码: {tenants_resp.status_code}")
print(f"响应: {json.dumps(tenants_resp.json(), indent=2, ensure_ascii=False)}")
# 3. 查看可用的平台 Agent 模板
print("\n" + "=" * 50)
print("3. 查看可用的平台 Agent 模板")
templates_resp = requests.get(
f"{BASE_URL}/api/channel/available-platform-agents",
headers=headers
)
print(f"状态码: {templates_resp.status_code}")
print(f"响应: {json.dumps(templates_resp.json(), indent=2, ensure_ascii=False)}")
# 4. 查看渠道的平台 Agent 配额
print("\n" + "=" * 50)
print("4. 查看渠道的平台 Agent 配额")
quotas_resp = requests.get(
f"{BASE_URL}/api/channel/platform-agents",
headers=headers
)
print(f"状态码: {quotas_resp.status_code}")
print(f"响应: {json.dumps(quotas_resp.json(), indent=2, ensure_ascii=False)}")
# 5. 如果有租户，尝试分配平台 Agent
if tenants_resp.status_code == 200:
tenants_data = tenants_resp.json()
tenants = tenants_data.get("data", {}).get("tenants", [])
if tenants:
tenant_id = tenants[0]["id"]
print("\n" + "=" * 50)
print(f"5. 分配平台 Agent 给租户 {tenant_id}")
allocate_resp = requests.post(
f"{BASE_URL}/api/channel/tenants/{tenant_id}/platform-agents",
headers=headers,
json={
"templateName": "gpt-assistant",
"podQuota": 2
}
)
print(f"状态码: {allocate_resp.status_code}")
print(f"响应: {json.dumps(allocate_resp.json(), indent=2, ensure_ascii=False)}")
else:
print("\n没有租户，跳过分配测试")
print("请先创建一个租户")
if __name__ == "__main__":
test_allocate_platform_agent()
test_channel_agent_workflow.sh
Executable
+447
View File
@@ -0,0 +1,447 @@
#!/bin/bash
# ============================================================================
# 渠道申请平台 Agent 和分配给租户的完整测试脚本
# ============================================================================
#
# 测试流程：
# 1. 创建超级管理员和渠道66
# 2. 超级管理员登录
# 3. 渠道66登录
# 4. 渠道申请平台 Agent
# 5. 超级管理员审批申请
# 6. 渠道创建租户
# 7. 渠道分配平台 Agent 给租户
#
# ============================================================================
set -e
BASE_URL="http://localhost:8002"
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# 打印函数
print_step() {
echo -e "\n${BLUE}========================================${NC}"
echo -e "${BLUE}步骤 $1: $2${NC}"
echo -e "${BLUE}========================================${NC}"
}
print_success() {
echo -e "${GREEN}✓ $1${NC}"
}
print_error() {
echo -e "${RED}✗ $1${NC}"
}
print_info() {
echo -e "${YELLOW}→ $1${NC}"
}
# ============================================================================
# 步骤 1: 创建超级管理员和渠道66（通过数据库脚本）
# ============================================================================
print_step "1" "创建超级管理员和渠道66"
# 运行 Python 脚本创建用户
python3 << 'EOF'
import sys
import os
import asyncio
import bcrypt
# 添加路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'services', 'mcp-server'))
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from models import User, Channel
from config import settings
def get_password_hash(password: str) -> str:
"""加密密码"""
password_bytes = password.encode('utf-8')
salt = bcrypt.gensalt()
hashed = bcrypt.hashpw(password_bytes, salt)
return hashed.decode('utf-8')
async def create_users():
"""创建超级管理员和渠道66"""
database_url = settings.database_url
if database_url.startswith("postgresql://") and "+asyncpg" not in database_url:
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
engine = create_async_engine(database_url, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with AsyncSessionLocal() as session:
# 1. 创建超级管理员
result = await session.execute(
select(User).where(User.email == "superadmin@taiji-ai.com")
)
existing_user = result.scalar_one_or_none()
if existing_user:
print(" ⚠ 超级管理员已存在，更新密码和角色")
password_hash = get_password_hash("Admin@123456")
existing_user.password_hash = password_hash
existing_user.hashed_password = password_hash
existing_user.role = "super_admin"
existing_user.is_admin = True
existing_user.is_active = True
else:
password_hash = get_password_hash("Admin@123456")
user = User(
name="超级管理员",
email="superadmin@taiji-ai.com",
password_hash=password_hash,
hashed_password=password_hash,
username="superadmin",
full_name="超级管理员",
role="super_admin",
is_active=True,
is_admin=True,
status="active",
balance=0,
credit_limit=0,
)
session.add(user)
print(" ✓ 创建超级管理员: superadmin@taiji-ai.com")
# 2. 创建渠道66
result = await session.execute(
select(Channel).where(Channel.email == "66@66.com")
)
existing_channel = result.scalar_one_or_none()
if existing_channel:
print(" ⚠ 渠道66已存在")
channel_id = str(existing_channel.id)
else:
password_hash = get_password_hash("66")
channel = Channel(
name="渠道66",
email="66@66.com",
password_hash=password_hash,
commission_rate=10.0,
channel_credit=100000,
custom_agent_cpu=8,
custom_agent_memory=16,
status="active",
)
session.add(channel)
await session.flush()
channel_id = str(channel.id)
print(f" ✓ 创建渠道66: 66@66.com (ID: {channel_id})")
# 3. 创建渠道管理员（关联到渠道66）
result = await session.execute(
select(Channel).where(Channel.email == "66@66.com")
)
channel = result.scalar_one_or_none()
result = await session.execute(
select(User).where(User.email == "channel66@taiji-ai.com")
)
existing_channel_admin = result.scalar_one_or_none()
if existing_channel_admin:
print(" ⚠ 渠道66管理员已存在，更新关联")
existing_channel_admin.channel_id = channel.id
existing_channel_admin.role = "channel_admin"
else:
password_hash = get_password_hash("Channel66@123")
channel_admin = User(
name="渠道66管理员",
email="channel66@taiji-ai.com",
password_hash=password_hash,
hashed_password=password_hash,
username="channel66",
full_name="渠道66管理员",
role="channel_admin",
channel_id=channel.id,
is_active=True,
is_admin=False,
status="active",
balance=0,
credit_limit=0,
)
session.add(channel_admin)
print(f" ✓ 创建渠道66管理员: channel66@taiji-ai.com")
await session.commit()
print(f"\n 渠道66 ID: {channel_id}")
# 输出渠道ID供后续使用
with open('/tmp/channel66_id.txt', 'w') as f:
f.write(channel_id)
await engine.dispose()
asyncio.run(create_users())
EOF
if [ $? -eq 0 ]; then
print_success "用户创建完成"
else
print_error "用户创建失败"
exit 1
fi
# 读取渠道ID
CHANNEL_ID=$(cat /tmp/channel66_id.txt 2>/dev/null || echo "")
print_info "渠道66 ID: $CHANNEL_ID"
# ============================================================================
# 步骤 2: 超级管理员登录
# ============================================================================
print_step "2" "超级管理员登录"
SUPER_ADMIN_RESPONSE=$(curl -s -X POST "$BASE_URL/api/auth/login" \
-H "Content-Type: application/json" \
-d '{
"email": "superadmin@taiji-ai.com",
"password": "Admin@123456",
"role": "super_admin"
}')
echo "响应: $SUPER_ADMIN_RESPONSE"
SUPER_ADMIN_TOKEN=$(echo $SUPER_ADMIN_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); print(data.get('data', {}).get('token', ''))" 2>/dev/null || echo "")
if [ -n "$SUPER_ADMIN_TOKEN" ]; then
print_success "超级管理员登录成功"
print_info "Token: ${SUPER_ADMIN_TOKEN:0:50}..."
else
print_error "超级管理员登录失败"
exit 1
fi
# ============================================================================
# 步骤 3: 渠道66管理员登录
# ============================================================================
print_step "3" "渠道66管理员登录"
CHANNEL_ADMIN_RESPONSE=$(curl -s -X POST "$BASE_URL/api/auth/login" \
-H "Content-Type: application/json" \
-d '{
"email": "channel66@taiji-ai.com",
"password": "Channel66@123",
"role": "channel_admin"
}')
echo "响应: $CHANNEL_ADMIN_RESPONSE"
CHANNEL_ADMIN_TOKEN=$(echo $CHANNEL_ADMIN_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); print(data.get('data', {}).get('token', ''))" 2>/dev/null || echo "")
if [ -n "$CHANNEL_ADMIN_TOKEN" ]; then
print_success "渠道66管理员登录成功"
print_info "Token: ${CHANNEL_ADMIN_TOKEN:0:50}..."
else
print_error "渠道66管理员登录失败"
exit 1
fi
# ============================================================================
# 步骤 4: 查看可用的平台 Agent 模板
# ============================================================================
print_step "4" "查看可用的平台 Agent 模板"
curl -s -X GET "$BASE_URL/api/channel/available-platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" | python3 -m json.tool
print_success "查看平台 Agent 模板完成"
# ============================================================================
# 步骤 5: 渠道申请平台 Agent (gpt-assistant)
# ============================================================================
print_step "5" "渠道申请平台 Agent (gpt-assistant)"
APPLICATION_RESPONSE=$(curl -s -X POST "$BASE_URL/api/channel/applications/platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"templateName": "gpt-assistant",
"requestedPodQuota": 5,
"reason": "业务需要使用 GPT 智能助手进行客户服务"
}')
echo "响应: $APPLICATION_RESPONSE" | python3 -m json.tool
APPLICATION_ID=$(echo $APPLICATION_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); print(data.get('data', {}).get('id', ''))" 2>/dev/null || echo "")
if [ -n "$APPLICATION_ID" ]; then
print_success "申请提交成功，申请ID: $APPLICATION_ID"
else
print_info "申请可能已存在或提交失败"
fi
# ============================================================================
# 步骤 6: 查看渠道的申请列表
# ============================================================================
print_step "6" "查看渠道的申请列表"
curl -s -X GET "$BASE_URL/api/channel/applications/platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" | python3 -m json.tool
# 获取待审批的申请ID
PENDING_APP_RESPONSE=$(curl -s -X GET "$BASE_URL/api/channel/applications/platform-agents?status=pending" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN")
APPLICATION_ID=$(echo $PENDING_APP_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); apps=data.get('data', {}).get('applications', []); print(apps[0]['id'] if apps else '')" 2>/dev/null || echo "")
print_info "待审批申请ID: $APPLICATION_ID"
# ============================================================================
# 步骤 7: 超级管理员查看所有申请
# ============================================================================
print_step "7" "超级管理员查看所有申请"
curl -s -X GET "$BASE_URL/api/admin/applications/platform-agents" \
-H "Authorization: Bearer $SUPER_ADMIN_TOKEN" | python3 -m json.tool
# ============================================================================
# 步骤 8: 超级管理员审批申请
# ============================================================================
print_step "8" "超级管理员审批申请"
if [ -n "$APPLICATION_ID" ]; then
REVIEW_RESPONSE=$(curl -s -X PUT "$BASE_URL/api/admin/applications/platform-agents/$APPLICATION_ID/review" \
-H "Authorization: Bearer $SUPER_ADMIN_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"approved": true,
"approvedPodQuota": 5,
"reason": "批准使用，配额5个Pod"
}')
echo "响应: $REVIEW_RESPONSE" | python3 -m json.tool
print_success "申请审批完成"
else
print_info "没有待审批的申请，尝试直接分配配额"
# 直接分配配额给渠道
ALLOCATE_RESPONSE=$(curl -s -X POST "$BASE_URL/api/admin/platform-agents/allocate?channel_id=$CHANNEL_ID&template_name=gpt-assistant&pod_quota=5" \
-H "Authorization: Bearer $SUPER_ADMIN_TOKEN")
echo "响应: $ALLOCATE_RESPONSE" | python3 -m json.tool
print_success "直接分配配额完成"
fi
# ============================================================================
# 步骤 9: 查看渠道的平台 Agent 配额
# ============================================================================
print_step "9" "查看渠道的平台 Agent 配额"
curl -s -X GET "$BASE_URL/api/channel/platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" | python3 -m json.tool
print_success "查看配额完成"
# ============================================================================
# 步骤 10: 渠道创建租户
# ============================================================================
print_step "10" "渠道创建租户"
TENANT_RESPONSE=$(curl -s -X POST "$BASE_URL/api/channel/tenants/create" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"name": "测试租户A",
"email": "tenant-a@test.com",
"password": "TenantA@123",
"subscriptionTier": "premium"
}')
echo "响应: $TENANT_RESPONSE" | python3 -m json.tool
TENANT_ID=$(echo $TENANT_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); print(data.get('data', {}).get('id', ''))" 2>/dev/null || echo "")
if [ -n "$TENANT_ID" ]; then
print_success "租户创建成功，租户ID: $TENANT_ID"
else
# 如果租户已存在，尝试获取租户ID
print_info "租户可能已存在，尝试获取租户列表"
TENANTS_RESPONSE=$(curl -s -X GET "$BASE_URL/api/channel/tenants" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN")
TENANT_ID=$(echo $TENANTS_RESPONSE | python3 -c "import sys, json; data=json.load(sys.stdin); tenants=data.get('data', {}).get('tenants', []); print(tenants[0]['id'] if tenants else '')" 2>/dev/null || echo "")
if [ -n "$TENANT_ID" ]; then
print_info "使用现有租户ID: $TENANT_ID"
fi
fi
# ============================================================================
# 步骤 11: 渠道分配平台 Agent 给租户
# ============================================================================
print_step "11" "渠道分配平台 Agent 给租户"
if [ -n "$TENANT_ID" ]; then
ALLOCATE_TENANT_RESPONSE=$(curl -s -X POST "$BASE_URL/api/channel/tenants/$TENANT_ID/platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"templateName": "gpt-assistant",
"podQuota": 3
}')
echo "响应: $ALLOCATE_TENANT_RESPONSE" | python3 -m json.tool
print_success "平台 Agent 分配给租户完成"
else
print_error "没有可用的租户ID"
fi
# ============================================================================
# 步骤 12: 查看租户的平台 Agent 使用情况
# ============================================================================
print_step "12" "查看租户的平台 Agent 使用情况"
if [ -n "$TENANT_ID" ]; then
curl -s -X GET "$BASE_URL/api/channel/tenants/$TENANT_ID/platform-agents/usage" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" | python3 -m json.tool
print_success "查看租户使用情况完成"
fi
# ============================================================================
# 步骤 13: 再次查看渠道的平台 Agent 配额（验证分配后的变化）
# ============================================================================
print_step "13" "再次查看渠道的平台 Agent 配额"
curl -s -X GET "$BASE_URL/api/channel/platform-agents" \
-H "Authorization: Bearer $CHANNEL_ADMIN_TOKEN" | python3 -m json.tool
# ============================================================================
# 测试完成
# ============================================================================
echo -e "\n${GREEN}============================================${NC}"
echo -e "${GREEN}测试完成！${NC}"
echo -e "${GREEN}============================================${NC}"
echo -e "\n${YELLOW}账户信息汇总：${NC}"
echo -e " 超级管理员: superadmin@taiji-ai.com / Admin@123456"
echo -e " 渠道66管理员: channel66@taiji-ai.com / Channel66@123"
echo -e " 渠道66 ID: $CHANNEL_ID"
if [ -n "$TENANT_ID" ]; then
echo -e " 测试租户ID: $TENANT_ID"
fi
echo -e "\n${YELLOW}测试流程：${NC}"
echo -e " 1. ✓ 创建超级管理员和渠道66"
echo -e " 2. ✓ 超级管理员登录"
echo -e " 3. ✓ 渠道66管理员登录"
echo -e " 4. ✓ 查看可用的平台 Agent 模板"
echo -e " 5. ✓ 渠道申请平台 Agent"
echo -e " 6. ✓ 查看渠道的申请列表"
echo -e " 7. ✓ 超级管理员查看所有申请"
echo -e " 8. ✓ 超级管理员审批申请"
echo -e " 9. ✓ 查看渠道的平台 Agent 配额"
echo -e " 10. ✓ 渠道创建租户"
echo -e " 11. ✓ 渠道分配平台 Agent 给租户"
echo -e " 12. ✓ 查看租户的平台 Agent 使用情况"
echo -e " 13. ✓ 再次查看渠道的平台 Agent 配额"