digital-employee-platform/backend/tests/test_employees.py

import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_create_employee(client: AsyncClient):
    """创建数字员工"""
    # 先创建租户
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "Employee Test", "slug": "employee-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    response = await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "客服助手",
            "role": "customer_service",
            "system_prompt": "你是一个专业的客服助手。",
            "greeting": "你好，有什么可以帮助您的？",
            "temperature": 0.7,
            "max_context_messages": 20,
            "knowledge_base_ids": [],
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "客服助手"
    assert data["role"] == "customer_service"
    assert data["status"] == "active"
    assert "id" in data


@pytest.mark.asyncio
async def test_create_employee_with_knowledge_bases(client: AsyncClient):
    """创建数字员工时关联知识库"""
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "KB Test", "slug": "kb-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    # 创建知识库
    from app.models import KnowledgeBase
    from tests.conftest import TestSession

    async with TestSession() as session:
        kb1 = KnowledgeBase(tenant_id=tenant_id, name="产品文档")
        kb2 = KnowledgeBase(tenant_id=tenant_id, name="FAQ")
        session.add_all([kb1, kb2])
        await session.commit()
        kb1_id = kb1.id
        kb2_id = kb2.id

    response = await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "智能客服",
            "role": "customer_service",
            "system_prompt": "你是一个智能客服。",
            "knowledge_base_ids": [kb1_id, kb2_id],
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert len(data["knowledge_base_ids"]) == 2


@pytest.mark.asyncio
async def test_list_employees(client: AsyncClient):
    """列出数字员工"""
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "List Test", "slug": "list-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "员工A",
            "role": "sales",
            "system_prompt": "销售助手",
        },
    )
    await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "员工B",
            "role": "support",
            "system_prompt": "支持助手",
        },
    )

    response = await client.get(f"/api/v1/employees?tenant_id={tenant_id}")
    assert response.status_code == 200
    data = response.json()
    assert len(data) >= 2


@pytest.mark.asyncio
async def test_get_employee(client: AsyncClient):
    """获取单个数字员工"""
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "Get Emp Test", "slug": "get-emp-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    create_resp = await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "测试员工",
            "role": "test",
            "system_prompt": "测试",
        },
    )
    employee_id = create_resp.json()["id"]

    response = await client.get(f"/api/v1/employees/{employee_id}")
    assert response.status_code == 200
    assert response.json()["name"] == "测试员工"


@pytest.mark.asyncio
async def test_update_employee(client: AsyncClient):
    """更新数字员工"""
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "Update Emp Test", "slug": "update-emp-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    create_resp = await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "旧名称",
            "role": "old_role",
            "system_prompt": "旧提示词",
        },
    )
    employee_id = create_resp.json()["id"]

    response = await client.patch(
        f"/api/v1/employees/{employee_id}",
        json={
            "name": "新名称",
            "temperature": 0.5,
        },
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "新名称"
    assert data["temperature"] == 0.5


@pytest.mark.asyncio
async def test_delete_employee(client: AsyncClient):
    """删除数字员工"""
    tenant_resp = await client.post(
        "/api/v1/tenants",
        json={"name": "Delete Emp Test", "slug": "delete-emp-test"},
    )
    tenant_id = tenant_resp.json()["id"]

    create_resp = await client.post(
        "/api/v1/employees",
        json={
            "tenant_id": tenant_id,
            "name": "待删除",
            "role": "temp",
            "system_prompt": "临时",
        },
    )
    employee_id = create_resp.json()["id"]

    response = await client.delete(f"/api/v1/employees/{employee_id}")
    assert response.status_code == 204

    get_resp = await client.get(f"/api/v1/employees/{employee_id}")
    assert get_resp.status_code == 404