219 lines
7.8 KiB
Python
219 lines
7.8 KiB
Python
|
# =============================================================================
|
||
|
|
# 企微IT智能服务台 — H5 摇人功能测试
|
||
|
|
# =============================================================================
|
||
|
|
# 测试覆盖：
|
||
|
|
# 1. 摇人成功（新建会话 + 举手标记 + 趣味话术返回）
|
||
|
|
# 2. 摇人成功（已有会话 + 更新举手标记）
|
||
|
|
# 3. 缺少 employee_id 请求失败
|
||
|
|
# 4. 获取当前会话
|
||
|
|
# 5. H5 发送消息
|
||
|
|
# 6. 审批链接获取
|
||
|
|
# 7. 软件下载列表获取
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import pytest_asyncio
|
||
|
|
from unittest.mock import AsyncMock, patch
|
||
|
|
|
||
|
|
from app.models.conversation import Conversation
|
||
|
|
from app.models.funny_phrase import FunnyPhrase
|
||
|
|
from tests.conftest import create_test_conversation, MockRedis
|
||
|
|
|
||
|
|
|
||
|
|
class TestShakeEndpoint:
|
||
|
|
"""测试摇人 API 端点。"""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_shake_creates_new_conversation(self, client, db_session):
|
||
|
|
"""验证摇人时如果没有活跃会话则创建新会话。"""
|
||
|
|
# 先添加趣味话术
|
||
|
|
phrase = FunnyPhrase(scene="shake", content="摇人话术测试", tone="亲切", sort_order=1)
|
||
|
|
db_session.add(phrase)
|
||
|
|
await db_session.flush()
|
||
|
|
|
||
|
|
response = await client.post(
|
||
|
|
"/h5/conversations/current/shake",
|
||
|
|
json={"employee_id": "shake_new_user", "employee_name": "测试员工"},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert data["data"]["conversation"]["tags"]["hand_raise"] is True
|
||
|
|
assert data["data"]["funny_phrase"] != ""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_shake_updates_existing_conversation(self, client, db_session):
|
||
|
|
"""验证摇人时如果已有活跃会话则更新举手标记。"""
|
||
|
|
conv = create_test_conversation(
|
||
|
|
employee_id="shake_existing_user",
|
||
|
|
status="queued",
|
||
|
|
tags={},
|
||
|
|
)
|
||
|
|
db_session.add(conv)
|
||
|
|
await db_session.flush()
|
||
|
|
|
||
|
|
# 添加话术
|
||
|
|
phrase = FunnyPhrase(scene="shake", content="更新摇人话术", tone="亲切", sort_order=1)
|
||
|
|
db_session.add(phrase)
|
||
|
|
await db_session.flush()
|
||
|
|
|
||
|
|
response = await client.post(
|
||
|
|
"/h5/conversations/current/shake",
|
||
|
|
json={"employee_id": "shake_existing_user", "employee_name": "已有用户"},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert data["data"]["conversation"]["tags"]["hand_raise"] is True
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_shake_returns_funny_phrase(self, client, db_session):
|
||
|
|
"""验证摇人返回趣味话术。"""
|
||
|
|
phrase = FunnyPhrase(scene="shake", content="测试趣味话术内容", tone="亲切", sort_order=1)
|
||
|
|
db_session.add(phrase)
|
||
|
|
await db_session.flush()
|
||
|
|
|
||
|
|
response = await client.post(
|
||
|
|
"/h5/conversations/current/shake",
|
||
|
|
json={"employee_id": "phrase_test_user", "employee_name": "话术测试"},
|
||
|
|
)
|
||
|
|
|
||
|
|
data = response.json()
|
||
|
|
assert data["data"]["funny_phrase"] != ""
|
||
|
|
|
||
|
|
|
||
|
|
class TestH5CurrentConversation:
|
||
|
|
"""测试 H5 获取当前会话。"""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_current_conversation_exists(self, client, db_session, mock_redis):
|
||
|
|
"""验证获取当前活跃会话。"""
|
||
|
|
# 预设 Bearer Token（替代旧的 X-Employee-Id 头）
|
||
|
|
await mock_redis.setex("employee:token:h5_current_token", 28800, "h5_current_user")
|
||
|
|
|
||
|
|
conv = create_test_conversation(employee_id="h5_current_user", status="queued")
|
||
|
|
db_session.add(conv)
|
||
|
|
await db_session.flush()
|
||
|
|
|
||
|
|
response = await client.get(
|
||
|
|
"/h5/conversations/current",
|
||
|
|
headers={"Authorization": "Bearer h5_current_token"},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert data["data"] is not None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_current_conversation_not_found(self, client, db_session, mock_redis):
|
||
|
|
"""验证无活跃会话时返回空数据。"""
|
||
|
|
# 预设 Bearer Token
|
||
|
|
await mock_redis.setex("employee:token:no_conv_token", 28800, "no_conversation_user")
|
||
|
|
|
||
|
|
response = await client.get(
|
||
|
|
"/h5/conversations/current",
|
||
|
|
headers={"Authorization": "Bearer no_conv_token"},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert data["data"] is None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_current_conversation_no_employee_id(self, client, db_session):
|
||
|
|
"""验证缺少员工ID时返回未授权错误。"""
|
||
|
|
response = await client.get("/h5/conversations/current")
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] != 0 # 应返回错误码
|
||
|
|
|
||
|
|
|
||
|
|
class TestH5SendMessage:
|
||
|
|
"""测试 H5 发送消息。"""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_message_creates_conversation(self, client, db_session, mock_redis):
|
||
|
|
"""验证发送消息时自动创建会话。"""
|
||
|
|
# 预设 Bearer Token
|
||
|
|
await mock_redis.setex("employee:token:h5_msg_token", 28800, "h5_msg_user")
|
||
|
|
|
||
|
|
response = await client.post(
|
||
|
|
"/h5/conversations/current/messages",
|
||
|
|
json={"content": "VPN连不上了"},
|
||
|
|
headers={"Authorization": "Bearer h5_msg_token"},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert data["data"]["content"] == "VPN连不上了"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_message_empty_content(self, client, db_session, mock_redis):
|
||
|
|
"""验证空消息内容返回错误。"""
|
||
|
|
# 预设 Bearer Token
|
||
|
|
await mock_redis.setex("employee:token:empty_msg_token", 28800, "empty_msg_user")
|
||
|
|
|
||
|
|
response = await client.post(
|
||
|
|
"/h5/conversations/current/messages",
|
||
|
|
json={"content": ""},
|
||
|
|
headers={"Authorization": "Bearer empty_msg_token"},
|
||
|
|
)
|
||
|
|
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] != 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestApprovalLinks:
|
||
|
|
"""测试审批链接获取。"""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_approval_links(self, client, seeded_db):
|
||
|
|
"""验证获取审批链接列表。"""
|
||
|
|
response = await client.get("/h5/approval-links")
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert len(data["data"]["items"]) > 0
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_approval_links_by_category(self, client, seeded_db):
|
||
|
|
"""验证按分类过滤审批链接。"""
|
||
|
|
response = await client.get("/h5/approval-links?category=IT")
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
for item in data["data"]["items"]:
|
||
|
|
assert item["category"] == "IT"
|
||
|
|
|
||
|
|
|
||
|
|
class TestSoftwareDownloads:
|
||
|
|
"""测试软件下载列表。"""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_software_downloads(self, client, seeded_db):
|
||
|
|
"""验证获取软件下载列表。"""
|
||
|
|
response = await client.get("/h5/software-downloads")
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
assert len(data["data"]["items"]) > 0
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_software_downloads_by_category(self, client, seeded_db):
|
||
|
|
"""验证按分类过滤软件下载。"""
|
||
|
|
response = await client.get("/h5/software-downloads?category=办公")
|
||
|
|
|
||
|
|
assert response.status_code == 200
|
||
|
|
data = response.json()
|
||
|
|
assert data["code"] == 0
|
||
|
|
for item in data["data"]["items"]:
|
||
|
|
assert item["category"] == "办公"
|