Simon
bea288e414
== 已部署上线 (9项) == - 代办事项真实数据源集成 (企微审批API 8bug修复链) - H5/坐席端 Logo样式统一+绿色背景 - 视频引导页修复 (localStorage key v2) - 坐席端 v9 Vue版本修复 (ElMessage._context) - 截图按钮 v10 修复 (getDisplayMedia user gesture) - 扫码样式恢复+H5扫码登录跳转修复 - H5截图快捷键提示 == 代码完成待部署 (3项) == - 知识迭代3Bug修复 (#8 POST端点/#7 MERGE幂等/#6 过期检查) - 会议室预定-小鱼易联终端 (40文件, 40/40测试通过) - IT资产升级审批推送 (asset_service.py) == 需求文档 (2项) == - 坐席端AI辅助消息框-PRD (4项新功能确认) - 坐席端布局优化建议 v2.0 (7天计划) == 新增文档 == - 日报-2026-07-11.md - 知识迭代Bug修复报告-20260711.md - 会议室预定-部署指南.md - CHANGELOG.md 更新 == 测试 == - test_todo_integration.py: 40/40 - test_meetingroom.py: 40/40 - test_bugfix_ki_suggestions.py: 21/21
835 lines
33 KiB
Python
835 lines
33 KiB
Python
# =============================================================================
|
||
# IT智能服务台 — 自备电脑补贴（BYOD）资格查询 API 测试
|
||
# =============================================================================
|
||
# 测试覆盖：
|
||
# 1. 岗位匹配逻辑 _match_position（精确/包含/关键词/不匹配/空字符串/全部13岗位）
|
||
# 2. 关键词预过滤 _byod_keyword_prefilter（命中/未命中/空字符串）
|
||
# 3. 关键词兜底 _byod_fallback_detect（命中/未命中/空字符串）
|
||
# 4. API 端点 /byod/detect-intent（关键词命中→fallback / 关键词未命中→prefilter）
|
||
# 5. API 端点 /byod/check-eligibility（空ID/有资格/无资格/企微异常/空岗位）
|
||
# 6. API 端点 /byod/eligible-positions（13岗位/结构/URL/notes）
|
||
# 7. 数据文件 byod_eligible_positions.json 验证
|
||
#
|
||
# 响应格式：{code: 0, data: {...}, message: "success"}（success_response 包装）
|
||
# =============================================================================
|
||
|
||
import json
|
||
import os
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.api.byod import (
|
||
BYOD_APPLICATION_URL,
|
||
BYOD_ELIGIBLE_POSITIONS,
|
||
BYOD_NOTES,
|
||
BYOD_POSITION_MATCH_KEYWORDS,
|
||
BYOD_PREFILTER_KEYWORDS,
|
||
_byod_fallback_detect,
|
||
_byod_keyword_prefilter,
|
||
_match_position,
|
||
)
|
||
from app.config import settings
|
||
|
||
|
||
# =============================================================================
|
||
# 辅助：收集全部 13 个资格岗位（用于参数化测试）
|
||
# =============================================================================
|
||
|
||
ALL_ELIGIBLE_POSITIONS: list[tuple[str, str, str]] = []
|
||
for _seq, _cats in BYOD_ELIGIBLE_POSITIONS.items():
|
||
for _cat, _positions in _cats.items():
|
||
for _pos in _positions:
|
||
ALL_ELIGIBLE_POSITIONS.append((_pos, _seq, _cat))
|
||
|
||
|
||
# =============================================================================
|
||
# 单元测试：_match_position
|
||
# =============================================================================
|
||
|
||
class TestMatchPosition:
|
||
"""测试岗位匹配逻辑 _match_position。"""
|
||
|
||
def test_exact_match(self):
|
||
"""精确匹配：position 与资格岗位完全一致 → 匹配成功。"""
|
||
matched, pos, category = _match_position("前端开发岗")
|
||
assert matched is True
|
||
assert pos == "前端开发岗"
|
||
assert category == "技术序列 - 开发类"
|
||
|
||
def test_contains_match(self):
|
||
"""包含匹配：资格岗位是员工岗位的子串 → 匹配成功。"""
|
||
matched, pos, category = _match_position("高级前端开发岗")
|
||
assert matched is True
|
||
assert pos == "前端开发岗"
|
||
assert category == "技术序列 - 开发类"
|
||
|
||
def test_keyword_match(self):
|
||
"""关键词匹配：员工岗位包含核心关键词 → 匹配成功。"""
|
||
# "前端工程师" 不包含 "前端开发岗" 也不包含 "前端开发"，
|
||
# 但包含关键词 "前端"
|
||
matched, pos, category = _match_position("前端工程师")
|
||
assert matched is True
|
||
assert pos == "前端开发岗"
|
||
assert category == "技术序列 - 开发类"
|
||
|
||
def test_no_match_it_support(self):
|
||
"""不匹配：IT支持组组长 → 匹配失败。"""
|
||
matched, pos, category = _match_position("IT支持组组长")
|
||
assert matched is False
|
||
assert pos == ""
|
||
assert category == ""
|
||
|
||
def test_no_match_sales(self):
|
||
"""不匹配：销售经理 → 匹配失败。"""
|
||
matched, pos, category = _match_position("销售经理")
|
||
assert matched is False
|
||
assert pos == ""
|
||
assert category == ""
|
||
|
||
def test_empty_string(self):
|
||
"""空字符串 → 匹配失败。"""
|
||
matched, pos, category = _match_position("")
|
||
assert matched is False
|
||
assert pos == ""
|
||
assert category == ""
|
||
|
||
def test_whitespace_only(self):
|
||
"""纯空白字符串 → 匹配失败。"""
|
||
matched, pos, category = _match_position(" ")
|
||
assert matched is False
|
||
assert pos == ""
|
||
assert category == ""
|
||
|
||
def test_stripped_match(self):
|
||
"""带前后空格的岗位 → 去空格后匹配成功。"""
|
||
matched, pos, category = _match_position(" 前端开发岗 ")
|
||
assert matched is True
|
||
assert pos == "前端开发岗"
|
||
assert category == "技术序列 - 开发类"
|
||
|
||
@pytest.mark.parametrize("position,sequence,category", ALL_ELIGIBLE_POSITIONS)
|
||
def test_all_eligible_positions_exact_match(self, position, sequence, category):
|
||
"""全部 13 个资格岗位逐一精确匹配测试。"""
|
||
matched, matched_pos, matched_cat = _match_position(position)
|
||
assert matched is True, f"岗位 '{position}' 应匹配但未匹配"
|
||
assert matched_pos == position
|
||
assert matched_cat == f"{sequence} - {category}"
|
||
|
||
def test_keyword_match_ux_design(self):
|
||
"""关键词匹配：用户体验设计岗的多种关键词变体。"""
|
||
# "UX设计师" 包含关键词 "UX设计"
|
||
matched, pos, _ = _match_position("UX设计师")
|
||
assert matched is True
|
||
assert pos == "用户体验设计岗"
|
||
|
||
# "交互设计专家" 包含关键词 "交互设计"
|
||
matched, pos, _ = _match_position("交互设计专家")
|
||
assert matched is True
|
||
assert pos == "用户体验设计岗"
|
||
|
||
def test_keyword_match_mobile(self):
|
||
"""关键词匹配：移动端开发岗的多种关键词变体。"""
|
||
# "移动开发组长" 包含关键词 "移动开发"
|
||
matched, pos, _ = _match_position("移动开发组长")
|
||
assert matched is True
|
||
assert pos == "移动端开发岗"
|
||
|
||
def test_no_match_partial_keyword(self):
|
||
"""不匹配：仅包含部分关键词（如"端"不匹配"客户端"）。"""
|
||
matched, _, _ = _match_position("服务端架构师")
|
||
assert matched is False
|
||
|
||
def test_return_type_is_tuple(self):
|
||
"""返回值类型为三元组 (bool, str, str)。"""
|
||
result = _match_position("前端开发岗")
|
||
assert isinstance(result, tuple)
|
||
assert len(result) == 3
|
||
assert isinstance(result[0], bool)
|
||
assert isinstance(result[1], str)
|
||
assert isinstance(result[2], str)
|
||
|
||
|
||
# =============================================================================
|
||
# 单元测试：_byod_keyword_prefilter
|
||
# =============================================================================
|
||
|
||
class TestByodKeywordPrefilter:
|
||
"""测试 BYOD 关键词预过滤函数。"""
|
||
|
||
def test_hit_zibei_diannao(self):
|
||
"""包含 '自备电脑' 关键词 → True。"""
|
||
assert _byod_keyword_prefilter("我想申请自备电脑补贴") is True
|
||
|
||
def test_hit_byod_uppercase(self):
|
||
"""包含 'BYOD' 关键词（大写）→ True。"""
|
||
assert _byod_keyword_prefilter("BYOD政策是什么") is True
|
||
|
||
def test_hit_byod_lowercase(self):
|
||
"""包含 'byod' 关键词（小写）→ True（大小写不敏感）。"""
|
||
assert _byod_keyword_prefilter("byod政策是什么") is True
|
||
|
||
def test_hit_diannao_butie(self):
|
||
"""包含 '电脑补贴' 关键词 → True。"""
|
||
assert _byod_keyword_prefilter("电脑补贴资格") is True
|
||
|
||
def test_hit_butie_zige(self):
|
||
"""包含 '补贴资格' 关键词 → True。"""
|
||
assert _byod_keyword_prefilter("我有补贴资格吗") is True
|
||
|
||
def test_miss_normal_message(self):
|
||
"""普通消息不包含 BYOD 关键词 → False。"""
|
||
assert _byod_keyword_prefilter("打印机坏了") is False
|
||
|
||
def test_miss_empty_string(self):
|
||
"""空字符串 → False。"""
|
||
assert _byod_keyword_prefilter("") is False
|
||
|
||
def test_miss_unrelated_it_message(self):
|
||
"""IT 相关但非 BYOD 的消息 → False。"""
|
||
assert _byod_keyword_prefilter("VPN连不上了") is False
|
||
assert _byod_keyword_prefilter("我的电脑黑屏了") is False
|
||
|
||
def test_case_insensitive(self):
|
||
"""关键词匹配大小写不敏感。"""
|
||
assert _byod_keyword_prefilter("Byod") is True
|
||
assert _byod_keyword_prefilter("BYOD") is True
|
||
assert _byod_keyword_prefilter("byod") is True
|
||
|
||
def test_all_prefilter_keywords_work(self):
|
||
"""验证 BYOD_PREFILTER_KEYWORDS 中的每个关键词都能被命中。"""
|
||
for kw in BYOD_PREFILTER_KEYWORDS:
|
||
text = f"测试文本包含{kw}关键词"
|
||
assert _byod_keyword_prefilter(text) is True, f"关键词 '{kw}' 未被命中"
|
||
|
||
|
||
# =============================================================================
|
||
# 单元测试：_byod_fallback_detect
|
||
# =============================================================================
|
||
|
||
class TestByodFallbackDetect:
|
||
"""测试 BYOD 关键词兜底检测函数。"""
|
||
|
||
def test_fallback_hit_returns_true(self):
|
||
"""包含 BYOD 关键词的文本 → 兜底返回 True。"""
|
||
is_byod, confidence = _byod_fallback_detect("我想申请自备电脑补贴")
|
||
assert is_byod is True
|
||
assert confidence == 0.6
|
||
|
||
def test_fallback_hit_byod_keyword(self):
|
||
"""包含 BYOD 关键词 → True。"""
|
||
is_byod, _ = _byod_fallback_detect("BYOD政策")
|
||
assert is_byod is True
|
||
|
||
def test_fallback_miss_normal_message(self):
|
||
"""普通消息不包含 BYOD 关键词 → False。"""
|
||
is_byod, confidence = _byod_fallback_detect("打印机坏了")
|
||
assert is_byod is False
|
||
assert confidence == 0.0
|
||
|
||
def test_fallback_empty_string(self):
|
||
"""空字符串 → False。"""
|
||
is_byod, confidence = _byod_fallback_detect("")
|
||
assert is_byod is False
|
||
assert confidence == 0.0
|
||
|
||
def test_fallback_none_text(self):
|
||
"""None → False（安全处理）。"""
|
||
is_byod, confidence = _byod_fallback_detect(None) # type: ignore[arg-type]
|
||
assert is_byod is False
|
||
assert confidence == 0.0
|
||
|
||
def test_fallback_confidence_is_0_6(self):
|
||
"""兜底置信度固定为 0.6。"""
|
||
_, confidence = _byod_fallback_detect("自备电脑补贴怎么申请")
|
||
assert confidence == 0.6
|
||
|
||
|
||
# =============================================================================
|
||
# API 端点测试：POST /byod/detect-intent
|
||
# =============================================================================
|
||
# 响应格式：{code: 0, data: {is_byod_intent, eligible, source, ...}, message: "success"}
|
||
# Dify 未配置时：关键词命中 → _call_dify_byod_intent 抛 ValueError → 降级兜底 source=fallback
|
||
# =============================================================================
|
||
|
||
class TestDetectByodIntentEndpoint:
|
||
"""测试 /byod/detect-intent API 端点。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_byod_keyword_hit_fallback(self, client):
|
||
"""包含 BYOD 关键词 + Dify 未配置 → is_byod_intent=True, source=fallback。"""
|
||
# Dify 未配置（approval_dify_base_url 为空），会触发降级兜底
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "我想申请自备电脑补贴"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 0
|
||
assert data["message"] == "success"
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is True
|
||
assert inner["source"] == "fallback"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_byod_keyword_hit_byod_text(self, client):
|
||
"""包含 'BYOD' 关键词 + Dify 未配置 → is_byod_intent=True, source=fallback。"""
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "BYOD政策是什么"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is True
|
||
assert inner["source"] == "fallback"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_keyword_not_hit_returns_false(self, client):
|
||
"""不包含 BYOD 关键词 → is_byod_intent=False, source=keyword_prefilter。"""
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "打印机坏了，帮我修一下"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 0
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is False
|
||
assert inner["source"] == "keyword_prefilter"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_text_returns_false(self, client):
|
||
"""空文本 → 关键词未命中, is_byod_intent=False。"""
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": ""},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is False
|
||
assert inner["source"] == "keyword_prefilter"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dify_configured_high_confidence(self, client, monkeypatch):
|
||
"""Dify 已配置 + 高置信度 → is_byod_intent=True, source=dify。"""
|
||
monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
|
||
monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
|
||
monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)
|
||
|
||
mock_dify = AsyncMock(return_value={
|
||
"is_byod_intent": True,
|
||
"confidence": 0.95,
|
||
})
|
||
|
||
with patch("app.api.byod._call_dify_byod_intent", mock_dify):
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "我想申请自备电脑补贴"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is True
|
||
assert inner["source"] == "dify"
|
||
mock_dify.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dify_failure_fallback(self, client, monkeypatch):
|
||
"""Dify 已配置但调用失败 → 降级兜底, source=fallback。"""
|
||
monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
|
||
monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
|
||
|
||
mock_dify = AsyncMock(side_effect=Exception("Dify 服务不可达"))
|
||
|
||
with patch("app.api.byod._call_dify_byod_intent", mock_dify):
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "自备电脑补贴怎么申请"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is True
|
||
assert inner["source"] == "fallback"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dify_low_confidence_returns_false(self, client, monkeypatch):
|
||
"""Dify 已配置 + 低置信度 → is_byod_intent=False, source=dify。"""
|
||
monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
|
||
monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
|
||
monkeypatch.setattr(settings, "approval_confidence_threshold", 0.7)
|
||
|
||
mock_dify = AsyncMock(return_value={
|
||
"is_byod_intent": True,
|
||
"confidence": 0.5,
|
||
})
|
||
|
||
with patch("app.api.byod._call_dify_byod_intent", mock_dify):
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "电脑补贴资格"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["is_byod_intent"] is False
|
||
assert inner["source"] == "dify"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_response_has_all_required_fields(self, client):
|
||
"""验证响应包含 ByodEligibilityResponse 的所有必需字段。"""
|
||
response = await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "你好"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert "is_byod_intent" in inner
|
||
assert "eligible" in inner
|
||
assert "position" in inner
|
||
assert "matched_category" in inner
|
||
assert "application_url" in inner
|
||
assert "notes" in inner
|
||
assert "reason" in inner
|
||
assert "source" in inner
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_employee_id_passed_to_dify(self, client, monkeypatch):
|
||
"""验证 employee_id 被传递给 Dify 调用。"""
|
||
monkeypatch.setattr(settings, "approval_dify_base_url", "http://test-dify")
|
||
monkeypatch.setattr(settings, "approval_dify_api_key", "test-api-key")
|
||
|
||
mock_dify = AsyncMock(return_value={
|
||
"is_byod_intent": True,
|
||
"confidence": 0.9,
|
||
})
|
||
|
||
with patch("app.api.byod._call_dify_byod_intent", mock_dify):
|
||
await client.post(
|
||
"/byod/detect-intent",
|
||
json={"text": "自备电脑补贴", "employee_id": "test_emp_001"},
|
||
)
|
||
|
||
mock_dify.assert_called_once_with("自备电脑补贴", "test_emp_001")
|
||
|
||
|
||
# =============================================================================
|
||
# API 端点测试：POST /byod/check-eligibility
|
||
# =============================================================================
|
||
# 需要 mock Redis（get_redis 从 app.main 导入 redis_client）和 WecomService
|
||
# conftest 已 patch app.services.wecom_service.WecomService，但 byod.py 在模块
|
||
# 加载时通过 `from app.services.wecom_service import WecomService` 获得了自己的
|
||
# 引用，因此必须额外 patch app.api.byod.WecomService 才能让 mock 生效。
|
||
# =============================================================================
|
||
|
||
@pytest.fixture
|
||
def byod_wecom_mock(mock_wecom_instance):
|
||
"""Patch app.api.byod.WecomService 并保存/恢复 get_user_info 的 mock 状态。
|
||
|
||
conftest 在模块级设置了 mock_wecom_module.get_user_info.side_effect，
|
||
BYOD 测试需要临时覆盖它以返回不同岗位，测试后恢复原值。
|
||
同时 patch byod 模块中的 WecomService 引用（conftest 未覆盖此模块）。
|
||
"""
|
||
original_se = mock_wecom_instance.get_user_info.side_effect
|
||
original_rv = mock_wecom_instance.get_user_info.return_value
|
||
# patch byod.py 模块中的 WecomService 引用（from ... import WecomService）
|
||
with patch("app.api.byod.WecomService", return_value=mock_wecom_instance):
|
||
yield mock_wecom_instance
|
||
mock_wecom_instance.get_user_info.side_effect = original_se
|
||
mock_wecom_instance.get_user_info.return_value = original_rv
|
||
|
||
|
||
class TestCheckEligibilityEndpoint:
|
||
"""测试 /byod/check-eligibility API 端点。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_employee_id(self, client, mock_redis):
|
||
"""employee_id 为空 → eligible=False, reason 包含 '缺少员工ID'。"""
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": ""},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 0
|
||
inner = data["data"]
|
||
assert inner["eligible"] is False
|
||
assert "缺少员工ID" in inner["reason"]
|
||
assert inner["source"] == "wecom"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_eligible_position(self, client, mock_redis, byod_wecom_mock):
|
||
"""岗位在资格清单中 → eligible=True, 返回申请链接和注意事项。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "前端开发岗",
|
||
"name": "张三",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "zhangsan"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 0
|
||
inner = data["data"]
|
||
assert inner["eligible"] is True
|
||
assert inner["position"] == "前端开发岗"
|
||
assert inner["matched_category"] == "技术序列 - 开发类"
|
||
assert inner["application_url"] == BYOD_APPLICATION_URL
|
||
assert inner["notes"] == BYOD_NOTES
|
||
assert inner["source"] == "wecom"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_eligible_position_contains_match(self, client, mock_redis, byod_wecom_mock):
|
||
"""包含匹配：岗位 '高级后端开发岗' → eligible=True。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "高级后端开发岗",
|
||
"name": "李四",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "lisi"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is True
|
||
assert inner["position"] == "高级后端开发岗"
|
||
assert inner["matched_category"] == "技术序列 - 开发类"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_eligible_position_keyword_match(self, client, mock_redis, byod_wecom_mock):
|
||
"""关键词匹配：岗位 '算法工程师' → eligible=True（关键词 '算法'）。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "算法工程师",
|
||
"name": "王五",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "wangwu"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is True
|
||
assert inner["matched_category"] == "技术序列 - 开发类"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ineligible_position(self, client, mock_redis, byod_wecom_mock):
|
||
"""岗位不在资格清单中 → eligible=False, reason 包含岗位名。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "销售经理",
|
||
"name": "赵六",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "zhaoliu"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is False
|
||
assert inner["position"] == "销售经理"
|
||
assert "销售经理" in inner["reason"]
|
||
assert inner["application_url"] == ""
|
||
assert inner["notes"] == []
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_wecom_service_failure(self, client, mock_redis, byod_wecom_mock):
|
||
"""企微 API 调用失败 → eligible=False, reason 包含错误信息。"""
|
||
async def _raise_error(user_id, **kwargs):
|
||
raise Exception("企微API不可达")
|
||
|
||
byod_wecom_mock.get_user_info.side_effect = _raise_error
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "test_emp"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is False
|
||
assert "获取员工信息失败" in inner["reason"]
|
||
assert inner["source"] == "wecom"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_position(self, client, mock_redis, byod_wecom_mock):
|
||
"""企微返回空岗位 → eligible=False, reason 提示联系IT服务台。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "",
|
||
"name": "测试员工",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "test_emp"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is False
|
||
assert "岗位信息" in inner["reason"]
|
||
assert inner["source"] == "wecom"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_product_manager_eligible(self, client, mock_redis, byod_wecom_mock):
|
||
"""产品经理岗位 → eligible=True（产品序列 - 产品策划与设计类）。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "产品经理",
|
||
"name": "产品",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "pm001"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["eligible"] is True
|
||
assert inner["matched_category"] == "产品序列 - 产品策划与设计类"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_response_has_all_required_fields(self, client, mock_redis, byod_wecom_mock):
|
||
"""验证响应包含所有必需字段。"""
|
||
byod_wecom_mock.get_user_info.side_effect = None
|
||
byod_wecom_mock.get_user_info.return_value = {
|
||
"position": "测试开发岗",
|
||
"name": "测试",
|
||
}
|
||
|
||
with patch("app.main.redis_client", mock_redis, create=True):
|
||
response = await client.post(
|
||
"/byod/check-eligibility",
|
||
json={"employee_id": "tester001"},
|
||
)
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert "is_byod_intent" in inner
|
||
assert "eligible" in inner
|
||
assert "position" in inner
|
||
assert "matched_category" in inner
|
||
assert "application_url" in inner
|
||
assert "notes" in inner
|
||
assert "reason" in inner
|
||
assert "source" in inner
|
||
|
||
|
||
# =============================================================================
|
||
# API 端点测试：GET /byod/eligible-positions
|
||
# =============================================================================
|
||
|
||
class TestEligiblePositionsEndpoint:
|
||
"""测试 /byod/eligible-positions API 端点。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_13_positions(self, client):
|
||
"""返回 13 个岗位, total_count=13。"""
|
||
response = await client.get("/byod/eligible-positions")
|
||
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 0
|
||
assert data["message"] == "success"
|
||
inner = data["data"]
|
||
assert inner["total_count"] == 13
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_positions_structure(self, client):
|
||
"""岗位清单结构正确：序列 → 类别 → 岗位列表。"""
|
||
response = await client.get("/byod/eligible-positions")
|
||
|
||
data = response.json()
|
||
positions = data["data"]["positions"]
|
||
|
||
# 顶级 key 应为序列名称
|
||
assert "技术序列" in positions
|
||
assert "产品序列" in positions
|
||
|
||
# 技术序列下有 3 个类别
|
||
tech = positions["技术序列"]
|
||
assert "开发类" in tech
|
||
assert "数据类" in tech
|
||
assert "测试类" in tech
|
||
|
||
# 开发类有 6 个岗位
|
||
assert len(tech["开发类"]) == 6
|
||
assert "前端开发岗" in tech["开发类"]
|
||
assert "算法岗" in tech["开发类"]
|
||
|
||
# 数据类有 2 个岗位
|
||
assert len(tech["数据类"]) == 2
|
||
|
||
# 测试类有 2 个岗位
|
||
assert len(tech["测试类"]) == 2
|
||
|
||
# 产品序列下有 1 个类别，3 个岗位
|
||
product = positions["产品序列"]
|
||
assert "产品策划与设计类" in product
|
||
assert len(product["产品策划与设计类"]) == 3
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_application_url_present(self, client):
|
||
"""返回 application_url 且不为空。"""
|
||
response = await client.get("/byod/eligible-positions")
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert inner["application_url"] == BYOD_APPLICATION_URL
|
||
assert "ehr.servyou.com.cn" in inner["application_url"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_notes_present(self, client):
|
||
"""返回 notes 列表且不为空。"""
|
||
response = await client.get("/byod/eligible-positions")
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
assert isinstance(inner["notes"], list)
|
||
assert len(inner["notes"]) == len(BYOD_NOTES)
|
||
assert len(inner["notes"]) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_total_count_matches_actual(self, client):
|
||
"""total_count 与实际岗位数一致。"""
|
||
response = await client.get("/byod/eligible-positions")
|
||
|
||
data = response.json()
|
||
inner = data["data"]
|
||
positions = inner["positions"]
|
||
actual_count = sum(
|
||
len(positions_list)
|
||
for categories in positions.values()
|
||
for positions_list in categories.values()
|
||
)
|
||
assert inner["total_count"] == actual_count
|
||
|
||
|
||
# =============================================================================
|
||
# 数据文件验证：data/byod_eligible_positions.json
|
||
# =============================================================================
|
||
|
||
class TestDataFileValidation:
|
||
"""验证 BYOD 资格岗位清单 JSON 数据文件。"""
|
||
|
||
# 数据文件路径（相对于 backend/ 目录）
|
||
DATA_FILE = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
|
||
"data",
|
||
"byod_eligible_positions.json",
|
||
)
|
||
|
||
@pytest.fixture
|
||
def data_content(self):
|
||
"""读取并解析数据文件。"""
|
||
with open(self.DATA_FILE, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
def test_file_exists(self):
|
||
"""数据文件存在。"""
|
||
assert os.path.exists(self.DATA_FILE), f"数据文件不存在: {self.DATA_FILE}"
|
||
|
||
def test_contains_13_positions(self, data_content):
|
||
"""数据文件包含 13 个岗位。"""
|
||
positions = data_content["positions"]
|
||
total = sum(
|
||
len(positions_list)
|
||
for categories in positions.values()
|
||
for positions_list in categories.values()
|
||
)
|
||
assert total == 13
|
||
assert data_content["total_count"] == 13
|
||
|
||
def test_structure_correct(self, data_content):
|
||
"""结构正确：序列 → 类别 → 岗位列表。"""
|
||
positions = data_content["positions"]
|
||
assert isinstance(positions, dict)
|
||
|
||
for seq_name, categories in positions.items():
|
||
assert isinstance(seq_name, str)
|
||
assert isinstance(categories, dict)
|
||
for cat_name, pos_list in categories.items():
|
||
assert isinstance(cat_name, str)
|
||
assert isinstance(pos_list, list)
|
||
for pos in pos_list:
|
||
assert isinstance(pos, str)
|
||
assert len(pos) > 0
|
||
|
||
def test_application_url_correct(self, data_content):
|
||
"""application_url 正确且指向 eHR 系统。"""
|
||
url = data_content["application_url"]
|
||
assert isinstance(url, str)
|
||
assert url.startswith("https://")
|
||
assert "ehr.servyou.com.cn" in url
|
||
assert "flowid=7747" in url
|
||
|
||
def test_notes_not_empty(self, data_content):
|
||
"""notes 不为空且为字符串列表。"""
|
||
notes = data_content["notes"]
|
||
assert isinstance(notes, list)
|
||
assert len(notes) > 0
|
||
for note in notes:
|
||
assert isinstance(note, str)
|
||
assert len(note) > 0
|
||
|
||
def test_data_matches_code(self, data_content):
|
||
"""数据文件中的岗位与代码中 BYOD_ELIGIBLE_POSITIONS 一致。"""
|
||
data_positions = data_content["positions"]
|
||
code_positions = BYOD_ELIGIBLE_POSITIONS
|
||
|
||
# 逐序列、逐类别、逐岗位对比
|
||
for seq, categories in code_positions.items():
|
||
assert seq in data_positions, f"序列 '{seq}' 在数据文件中不存在"
|
||
for cat, pos_list in categories.items():
|
||
assert cat in data_positions[seq], f"类别 '{seq}/{cat}' 在数据文件中不存在"
|
||
for pos in pos_list:
|
||
assert pos in data_positions[seq][cat], (
|
||
f"岗位 '{pos}' 在数据文件中不存在"
|
||
)
|
||
|
||
def test_all_positions_have_keywords(self):
|
||
"""所有资格岗位都在 BYOD_POSITION_MATCH_KEYWORDS 中有对应关键词。"""
|
||
for seq, categories in BYOD_ELIGIBLE_POSITIONS.items():
|
||
for cat, pos_list in categories.items():
|
||
for pos in pos_list:
|
||
assert pos in BYOD_POSITION_MATCH_KEYWORDS, (
|
||
f"岗位 '{pos}' 在 BYOD_POSITION_MATCH_KEYWORDS 中没有对应关键词"
|
||
)
|
||
assert len(BYOD_POSITION_MATCH_KEYWORDS[pos]) > 0, (
|
||
f"岗位 '{pos}' 的关键词列表为空"
|
||
)
|