c62156af53
B1: 项目脚手架 + 数据模型 + 租户管理 - Task 1.1: FastAPI 项目脚手架、SQLite + async SQLAlchemy - Task 1.2: 7 个数据模型 (Tenant, TenantConfig, DigitalEmployee, Conversation, Message, KnowledgeBase, Document) - Task 1.3: 租户 CRUD API + LLM 配置（含 API Key AES 加密） B2: 数字员工配置 + LLM Provider 抽象层 - Task 2.1: 数字员工 CRUD API（关联知识库） - Task 2.2: BaseLLMProvider 抽象接口 + OpenAI/Qwen Provider - Task 2.3: Provider 动态实例化 + test-provider 端点 验证: 26 个测试全部通过 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.models import Tenant
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_tenant(client: AsyncClient):
|
|
response = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "Test Company", "slug": "test-company"},
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["name"] == "Test Company"
|
|
assert data["slug"] == "test-company"
|
|
assert "id" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tenants(client: AsyncClient):
|
|
await client.post("/api/v1/tenants", json={"name": "A", "slug": "a"})
|
|
await client.post("/api/v1/tenants", json={"name": "B", "slug": "b"})
|
|
|
|
response = await client.get("/api/v1/tenants")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tenant(client: AsyncClient):
|
|
create_resp = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "Get Test", "slug": "get-test"},
|
|
)
|
|
tenant_id = create_resp.json()["id"]
|
|
|
|
response = await client.get(f"/api/v1/tenants/{tenant_id}")
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Get Test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_tenant(client: AsyncClient):
|
|
create_resp = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "Old Name", "slug": "old-slug"},
|
|
)
|
|
tenant_id = create_resp.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/tenants/{tenant_id}",
|
|
json={"name": "New Name"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "New Name"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_tenant(client: AsyncClient):
|
|
create_resp = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "To Delete", "slug": "to-delete"},
|
|
)
|
|
tenant_id = create_resp.json()["id"]
|
|
|
|
response = await client.delete(f"/api/v1/tenants/{tenant_id}")
|
|
assert response.status_code == 204
|
|
|
|
get_resp = await client.get(f"/api/v1/tenants/{tenant_id}")
|
|
assert get_resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_tenant_config(client: AsyncClient):
|
|
create_resp = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "Config Test", "slug": "config-test"},
|
|
)
|
|
tenant_id = create_resp.json()["id"]
|
|
|
|
response = await client.put(
|
|
f"/api/v1/tenants/{tenant_id}/config",
|
|
json={
|
|
"llm_provider": "openai",
|
|
"llm_api_key": "sk-test-key-12345",
|
|
"llm_model": "gpt-4",
|
|
"llm_base_url": None,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["llm_provider"] == "openai"
|
|
assert data["llm_model"] == "gpt-4"
|
|
assert "llm_api_key" not in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_key_encrypted(client: AsyncClient):
|
|
from tests.conftest import TestSession
|
|
|
|
create_resp = await client.post(
|
|
"/api/v1/tenants",
|
|
json={"name": "Encrypt Test", "slug": "encrypt-test"},
|
|
)
|
|
tenant_id = create_resp.json()["id"]
|
|
|
|
await client.put(
|
|
f"/api/v1/tenants/{tenant_id}/config",
|
|
json={
|
|
"llm_provider": "openai",
|
|
"llm_api_key": "sk-secret-key-99999",
|
|
"llm_model": "gpt-4",
|
|
},
|
|
)
|
|
|
|
async with TestSession() as session:
|
|
from sqlalchemy import select
|
|
from app.models import TenantConfig
|
|
|
|
result = await session.execute(
|
|
select(TenantConfig).where(TenantConfig.tenant_id == tenant_id)
|
|
)
|
|
config = result.scalar_one()
|
|
assert config.llm_api_key != "sk-secret-key-99999"
|
|
assert len(config.llm_api_key) > 20
|