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
874 lines
34 KiB
Python
874 lines
34 KiB
Python
# =============================================================================
|
||
# IT智能服务台 — 会议室预定功能测试
|
||
# =============================================================================
|
||
# 测试覆盖：
|
||
# MeetingroomService (mock WecomService + MockRedis):
|
||
# 1. get_room_list — 缓存命中/未命中
|
||
# 2. get_room_status — 预定状态查询
|
||
# 3. get_current_status — 实时状态（free/busy/starting_soon）
|
||
# 4. book_room — 预定成功后缓存清除
|
||
# 5. book_room — 时间冲突异常
|
||
# 6. cancel_booking — 取消成功后缓存清除
|
||
# 7. invalidate_room_cache — 缓存清除
|
||
# 8. get_booking_detail — 预定详情缓存
|
||
#
|
||
# API端点 (mock _get_meetingroom_service):
|
||
# 9. GET /itportal/meetingroom/list
|
||
# 10. GET /itportal/meetingroom/{id}/booking
|
||
# 11. GET /itportal/meetingroom/{id}/status
|
||
# 12. POST /itportal/meetingroom/book
|
||
# 13. DELETE /itportal/meetingroom/booking/{id}
|
||
# 14. GET /itportal/meetingroom/terminal/{sn}/binding
|
||
#
|
||
# 终端绑定CRUD (db_session + mock auth):
|
||
# 15. POST /itportal/admin/terminal-bindings — 新增
|
||
# 16. GET /itportal/admin/terminal-bindings — 列表
|
||
# 17. PUT /itportal/admin/terminal-bindings/{id} — 更新
|
||
# 18. DELETE /itportal/admin/terminal-bindings/{id} — 删除
|
||
#
|
||
# 模型:
|
||
# 19. TerminalRoomBinding — 字段定义
|
||
# 20. MeetingroomBookingSnapshot — 字段定义
|
||
# =============================================================================
|
||
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
|
||
from app.services.meetingroom_service import MeetingroomService
|
||
from app.models.terminal_room_binding import TerminalRoomBinding
|
||
from app.models.meetingroom_booking_snapshot import MeetingroomBookingSnapshot
|
||
from app.schemas.meetingroom import (
|
||
BookRequest,
|
||
TerminalBindingCreate,
|
||
TerminalBindingUpdate,
|
||
)
|
||
from app.dependencies import UserInfo
|
||
|
||
|
||
# =============================================================================
|
||
# 测试数据工厂
|
||
# =============================================================================
|
||
|
||
def make_room(id: int = 1, name: str = "测试会议室") -> dict:
|
||
"""构造企微API返回的会议室数据"""
|
||
return {
|
||
"meetingroom_id": id,
|
||
"name": name,
|
||
"capacity": 10,
|
||
"location": "18F东区",
|
||
"equipment": [1, 2],
|
||
"need_approval": 0,
|
||
}
|
||
|
||
|
||
def make_booking(
|
||
booking_id: str = "bk001",
|
||
subject: str = "项目周会",
|
||
start_offset_min: int = -30,
|
||
duration_min: int = 60,
|
||
status: int = 0,
|
||
) -> dict:
|
||
"""构造企微API返回的预定记录（start_offset_min 相对当前时间的偏移分钟数）"""
|
||
now = datetime.now()
|
||
start = now + timedelta(minutes=start_offset_min)
|
||
end = start + timedelta(minutes=duration_min)
|
||
return {
|
||
"booking_id": booking_id,
|
||
"subject": subject,
|
||
"booker": "zhangsan",
|
||
"booker_name": "张三",
|
||
"start_time": int(start.timestamp()),
|
||
"end_time": int(end.timestamp()),
|
||
"status": status,
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# MockRedis — 复用 conftest 中的 MockRedis 实现
|
||
# =============================================================================
|
||
|
||
class MockRedisHelper:
|
||
"""轻量级 MockRedis，用于 service 层测试（独立于 conftest）"""
|
||
|
||
def __init__(self):
|
||
self._data: dict = {}
|
||
|
||
async def get(self, key: str):
|
||
v = self._data.get(key)
|
||
if v is not None:
|
||
return v.encode("utf-8") if isinstance(v, str) else v
|
||
return None
|
||
|
||
async def setex(self, name: str, time: int, value: str):
|
||
self._data[name] = value
|
||
|
||
async def set(self, name: str, value: str, **kwargs):
|
||
self._data[name] = value
|
||
|
||
async def delete(self, *names):
|
||
count = 0
|
||
for n in names:
|
||
if n in self._data:
|
||
del self._data[n]
|
||
count += 1
|
||
return count
|
||
|
||
async def exists(self, *keys):
|
||
return sum(1 for k in keys if k in self._data)
|
||
|
||
async def expire(self, name: str, time: int):
|
||
return name in self._data
|
||
|
||
async def close(self):
|
||
pass
|
||
|
||
|
||
# =============================================================================
|
||
# MeetingroomService 单元测试
|
||
# =============================================================================
|
||
|
||
class TestMeetingroomService:
|
||
"""MeetingroomService 业务逻辑测试"""
|
||
|
||
@pytest.fixture
|
||
def mock_wecom(self):
|
||
"""创建 mock WecomService"""
|
||
wecom = AsyncMock()
|
||
wecom.get_meetingroom_list.return_value = [make_room(1, "A会议室"), make_room(2, "B会议室")]
|
||
wecom.get_booking_info.return_value = []
|
||
wecom.book_meetingroom.return_value = {"booking_id": "bk_new_001"}
|
||
wecom.cancel_booking.return_value = {"errcode": 0}
|
||
wecom.get_booking_detail.return_value = {
|
||
"booking_id": "bk001",
|
||
"subject": "详情测试",
|
||
"booker": "zhangsan",
|
||
}
|
||
return wecom
|
||
|
||
@pytest.fixture
|
||
def mock_redis(self):
|
||
return MockRedisHelper()
|
||
|
||
@pytest.fixture
|
||
def service(self, mock_wecom, mock_redis):
|
||
return MeetingroomService(wecom_service=mock_wecom, redis_client=mock_redis)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# get_room_list
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_room_list_with_cache(self, service, mock_wecom, mock_redis):
|
||
"""缓存命中时不调用企微API"""
|
||
# 第一次调用 — 应调用API并缓存
|
||
result1 = await service.get_room_list()
|
||
assert len(result1) == 2
|
||
assert mock_wecom.get_meetingroom_list.call_count == 1
|
||
|
||
# 第二次调用 — 应命中缓存，不调用API
|
||
result2 = await service.get_room_list()
|
||
assert len(result2) == 2
|
||
assert mock_wecom.get_meetingroom_list.call_count == 1 # 仍是1次
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_room_list_no_redis(self, mock_wecom):
|
||
"""无Redis时直接调用API"""
|
||
service = MeetingroomService(wecom_service=mock_wecom, redis_client=None)
|
||
result = await service.get_room_list()
|
||
assert len(result) == 2
|
||
assert mock_wecom.get_meetingroom_list.call_count == 1
|
||
|
||
# --------------------------------------------------------------------------
|
||
# get_room_status
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_room_status(self, service, mock_wecom):
|
||
"""验证预定状态查询"""
|
||
mock_wecom.get_booking_info.return_value = [
|
||
make_booking("bk001", "会议A", start_offset_min=60),
|
||
make_booking("bk002", "会议B", start_offset_min=120),
|
||
]
|
||
result = await service.get_room_status(meetingroom_id=1, date="2026-07-15")
|
||
assert len(result) == 2
|
||
assert result[0]["booking_id"] == "bk001"
|
||
assert mock_wecom.get_booking_info.call_count == 1
|
||
|
||
# 验证传给企微API的时间范围参数
|
||
call_args = mock_wecom.get_booking_info.call_args
|
||
assert call_args[0][0] == 1 # meetingroom_id
|
||
assert "2026-07-15T00:00:00" in call_args[0][1] # start_time
|
||
assert "2026-07-15T23:59:59" in call_args[0][2] # end_time
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_room_status_default_date(self, service, mock_wecom):
|
||
"""未传date时默认今天"""
|
||
await service.get_room_status(meetingroom_id=1)
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
call_args = mock_wecom.get_booking_info.call_args
|
||
assert today in call_args[0][1]
|
||
|
||
# --------------------------------------------------------------------------
|
||
# get_current_status
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status_free(self, service, mock_wecom):
|
||
"""当前空闲状态判断"""
|
||
# 预定在未来2小时，不在15分钟内 → free
|
||
mock_wecom.get_booking_info.return_value = [
|
||
make_booking("bk001", "未来会议", start_offset_min=120, duration_min=60),
|
||
]
|
||
result = await service.get_current_status(meetingroom_id=1)
|
||
assert result["status"] == "free"
|
||
assert result["current_meeting"] is None
|
||
assert result["next_meeting"] is not None
|
||
assert result["minutes_to_next"] is not None
|
||
assert result["minutes_to_next"] > 15
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status_busy(self, service, mock_wecom):
|
||
"""当前使用中状态判断"""
|
||
# 预定在当前时间段内 → busy
|
||
mock_wecom.get_booking_info.return_value = [
|
||
make_booking("bk001", "进行中会议", start_offset_min=-10, duration_min=30),
|
||
]
|
||
result = await service.get_current_status(meetingroom_id=1)
|
||
assert result["status"] == "busy"
|
||
assert result["current_meeting"] is not None
|
||
assert result["current_meeting"]["subject"] == "进行中会议"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status_starting_soon(self, service, mock_wecom):
|
||
"""即将开始状态判断（15分钟内）"""
|
||
# 预定在5分钟后开始 → starting_soon
|
||
mock_wecom.get_booking_info.return_value = [
|
||
make_booking("bk001", "即将开始", start_offset_min=5, duration_min=30),
|
||
]
|
||
result = await service.get_current_status(meetingroom_id=1)
|
||
assert result["status"] == "starting_soon"
|
||
assert result["next_meeting"] is not None
|
||
assert result["minutes_to_next"] <= 15
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status_no_bookings(self, service, mock_wecom):
|
||
"""无预定时返回free"""
|
||
mock_wecom.get_booking_info.return_value = []
|
||
result = await service.get_current_status(meetingroom_id=1)
|
||
assert result["status"] == "free"
|
||
assert result["current_meeting"] is None
|
||
assert result["next_meeting"] is None
|
||
assert result["minutes_to_next"] is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status_skip_cancelled(self, service, mock_wecom):
|
||
"""跳过已取消的预定"""
|
||
mock_wecom.get_booking_info.return_value = [
|
||
make_booking("bk001", "已取消", start_offset_min=-10, duration_min=30, status=1),
|
||
]
|
||
result = await service.get_current_status(meetingroom_id=1)
|
||
assert result["status"] == "free"
|
||
|
||
# --------------------------------------------------------------------------
|
||
# book_room
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_book_room_success(self, service, mock_wecom, mock_redis):
|
||
"""预定成功后缓存被清除"""
|
||
# 先写入缓存
|
||
cache_key = service.CACHE_KEY_BOOKING_INFO.format(room_id=1, date=datetime.now().strftime("%Y-%m-%d"))
|
||
await mock_redis.setex(cache_key, 30, json.dumps([{"old": True}]))
|
||
|
||
# 执行预定
|
||
result = await service.book_room(
|
||
meetingroom_id=1,
|
||
subject="测试预定",
|
||
start_time="2026-07-15T10:00:00+08:00",
|
||
end_time="2026-07-15T11:00:00+08:00",
|
||
booker="zhangsan",
|
||
)
|
||
assert result["booking_id"] == "bk_new_001"
|
||
assert mock_wecom.book_meetingroom.call_count == 1
|
||
|
||
# 验证缓存被清除
|
||
cached = await mock_redis.get(cache_key)
|
||
assert cached is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_book_room_conflict(self, service, mock_wecom):
|
||
"""时间冲突处理 — 企微API抛异常时透传"""
|
||
mock_wecom.book_meetingroom.side_effect = Exception("时间冲突: 该时段已被预定")
|
||
|
||
with pytest.raises(Exception, match="时间冲突"):
|
||
await service.book_room(
|
||
meetingroom_id=1,
|
||
subject="冲突预定",
|
||
start_time="2026-07-15T10:00:00+08:00",
|
||
end_time="2026-07-15T11:00:00+08:00",
|
||
booker="zhangsan",
|
||
)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# cancel_booking
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancel_booking_success(self, service, mock_wecom, mock_redis):
|
||
"""取消成功后缓存被清除"""
|
||
# 先写入状态缓存
|
||
status_key = service.CACHE_KEY_STATUS.format(room_id=1)
|
||
await mock_redis.setex(status_key, 10, json.dumps({"status": "busy"}))
|
||
|
||
# 执行取消
|
||
result = await service.cancel_booking(booking_id="bk001", meetingroom_id=1)
|
||
assert result["errcode"] == 0
|
||
assert mock_wecom.cancel_booking.call_count == 1
|
||
|
||
# 验证缓存被清除
|
||
cached = await mock_redis.get(status_key)
|
||
assert cached is None
|
||
|
||
# --------------------------------------------------------------------------
|
||
# invalidate_room_cache
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_invalidate_room_cache(self, service, mock_redis):
|
||
"""缓存清除"""
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
booking_key = service.CACHE_KEY_BOOKING_INFO.format(room_id=1, date=today)
|
||
status_key = service.CACHE_KEY_STATUS.format(room_id=1)
|
||
|
||
# 写入缓存
|
||
await mock_redis.setex(booking_key, 30, json.dumps([{"test": True}]))
|
||
await mock_redis.setex(status_key, 10, json.dumps({"status": "busy"}))
|
||
|
||
# 清除缓存
|
||
await service.invalidate_room_cache(meetingroom_id=1)
|
||
|
||
# 验证两个缓存key都被删除
|
||
assert await mock_redis.get(booking_key) is None
|
||
assert await mock_redis.get(status_key) is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_invalidate_room_cache_no_redis(self, mock_wecom):
|
||
"""无Redis时invalidate不报错"""
|
||
service = MeetingroomService(wecom_service=mock_wecom, redis_client=None)
|
||
await service.invalidate_room_cache(meetingroom_id=1) # 不应抛异常
|
||
|
||
# --------------------------------------------------------------------------
|
||
# get_booking_detail
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_booking_detail_with_cache(self, service, mock_wecom, mock_redis):
|
||
"""预定详情缓存"""
|
||
result1 = await service.get_booking_detail(meetingroom_id=1, booking_id="bk001")
|
||
assert result1["booking_id"] == "bk001"
|
||
assert mock_wecom.get_booking_detail.call_count == 1
|
||
|
||
# 第二次应命中缓存
|
||
result2 = await service.get_booking_detail(meetingroom_id=1, booking_id="bk001")
|
||
assert result2["booking_id"] == "bk001"
|
||
assert mock_wecom.get_booking_detail.call_count == 1 # 仍1次
|
||
|
||
# --------------------------------------------------------------------------
|
||
# _parse_booking_time
|
||
# --------------------------------------------------------------------------
|
||
|
||
def test_parse_booking_time_timestamp(self):
|
||
"""解析时间戳格式"""
|
||
ts = int(datetime(2026, 7, 15, 10, 0, 0).timestamp())
|
||
result = MeetingroomService._parse_booking_time(ts)
|
||
assert result is not None
|
||
assert result.year == 2026
|
||
assert result.month == 7
|
||
assert result.day == 15
|
||
|
||
def test_parse_booking_time_iso(self):
|
||
"""解析ISO字符串格式"""
|
||
result = MeetingroomService._parse_booking_time("2026-07-15T10:00:00")
|
||
assert result is not None
|
||
assert result.year == 2026
|
||
|
||
def test_parse_booking_time_none(self):
|
||
"""None输入返回None"""
|
||
assert MeetingroomService._parse_booking_time(None) is None
|
||
|
||
def test_parse_booking_time_invalid(self):
|
||
"""无效输入返回None"""
|
||
assert MeetingroomService._parse_booking_time("invalid") is None
|
||
|
||
|
||
# =============================================================================
|
||
# API 端点测试 — 会议室预定
|
||
# =============================================================================
|
||
|
||
class TestMeetingroomAPI:
|
||
"""会议室预定API端点测试"""
|
||
|
||
@pytest.fixture
|
||
def mock_service(self):
|
||
"""创建 mock MeetingroomService"""
|
||
svc = AsyncMock(spec=MeetingroomService)
|
||
svc.get_room_list.return_value = [make_room(1, "A会议室"), make_room(2, "B会议室")]
|
||
svc.get_room_status.return_value = [
|
||
make_booking("bk001", "会议A", start_offset_min=60),
|
||
]
|
||
svc.get_current_status.return_value = {
|
||
"status": "free",
|
||
"current_meeting": None,
|
||
"next_meeting": None,
|
||
"minutes_to_next": None,
|
||
"bookings": [],
|
||
}
|
||
svc.book_room.return_value = {"booking_id": "bk_new_001"}
|
||
svc.cancel_booking.return_value = {"errcode": 0}
|
||
svc.get_booking_detail.return_value = {
|
||
"booking_id": "bk001",
|
||
"subject": "详情",
|
||
"booker": "zhangsan",
|
||
"booker_name": "张三",
|
||
"attendees": ["lisi"],
|
||
"start_time": "2026-07-15T10:00:00",
|
||
"end_time": "2026-07-15T11:00:00",
|
||
}
|
||
svc._format_booking = MeetingroomService._format_booking
|
||
return svc
|
||
|
||
@pytest_asyncio.fixture
|
||
async def meetingroom_client(self, mock_service, db_session):
|
||
"""创建测试客户端，mock掉 _get_meetingroom_service。
|
||
|
||
使用独立 FastAPI app（只挂载 meetingroom 路由），
|
||
避免全量 create_app() 触发 itsm_service.py 中 httpx.Timeout 兼容性问题。
|
||
"""
|
||
from httpx import ASGITransport, AsyncClient
|
||
from fastapi import FastAPI
|
||
from app.database import get_db
|
||
from app.utils.response import AppException, app_exception_handler
|
||
|
||
app = FastAPI()
|
||
# 注册全局异常处理器（让 AppException 返回 JSON 而非 500）
|
||
app.add_exception_handler(AppException, app_exception_handler)
|
||
# 只挂载会议室预定路由
|
||
from app.api.meetingroom import router as meetingroom_router
|
||
app.include_router(meetingroom_router)
|
||
|
||
async def _override_db():
|
||
yield db_session
|
||
|
||
app.dependency_overrides[get_db] = _override_db
|
||
|
||
with patch("app.api.meetingroom._get_meetingroom_service", return_value=mock_service):
|
||
with patch("app.api.meetingroom.WecomService", return_value=AsyncMock()):
|
||
with patch("app.services.wecom_service.WecomService", return_value=AsyncMock()):
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
yield ac
|
||
|
||
app.dependency_overrides.clear()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_meetingroom_list(self, meetingroom_client, mock_service):
|
||
"""GET /itportal/meetingroom/list"""
|
||
resp = await meetingroom_client.get("/itportal/meetingroom/list")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert len(data["data"]["rooms"]) == 2
|
||
assert data["data"]["rooms"][0]["name"] == "A会议室"
|
||
mock_service.get_room_list.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_booking_info(self, meetingroom_client, mock_service):
|
||
"""GET /itportal/meetingroom/{id}/booking"""
|
||
resp = await meetingroom_client.get("/itportal/meetingroom/1/booking")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert len(data["data"]["bookings"]) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_current_status(self, meetingroom_client, mock_service):
|
||
"""GET /itportal/meetingroom/{id}/status"""
|
||
resp = await meetingroom_client.get("/itportal/meetingroom/1/status")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert data["data"]["status"] == "free"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_terminal_binding_not_found(self, meetingroom_client):
|
||
"""GET /itportal/meetingroom/terminal/{sn}/binding — 未绑定时返回null"""
|
||
resp = await meetingroom_client.get("/itportal/meetingroom/terminal/UNKNOWN_SN/binding")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert data["data"] is None
|
||
|
||
|
||
# =============================================================================
|
||
# API 端点测试 — 终端绑定CRUD
|
||
# =============================================================================
|
||
|
||
class TestTerminalBindingCRUD:
|
||
"""终端绑定管理CRUD测试"""
|
||
|
||
@pytest_asyncio.fixture
|
||
async def admin_client(self, db_session):
|
||
"""创建带admin权限的测试客户端。
|
||
|
||
使用独立 FastAPI app（只挂载 terminal_binding 路由），
|
||
避免全量 create_app() 触发 itsm_service.py 中 httpx.Timeout 兼容性问题。
|
||
"""
|
||
from httpx import ASGITransport, AsyncClient
|
||
from fastapi import FastAPI
|
||
from app.database import get_db
|
||
from app.dependencies import get_current_user
|
||
from app.utils.response import AppException, app_exception_handler
|
||
|
||
app = FastAPI()
|
||
# 注册全局异常处理器（让 AppException 返回 JSON 而非 500）
|
||
app.add_exception_handler(AppException, app_exception_handler)
|
||
from app.api.admin.terminal_binding import router as terminal_binding_router
|
||
app.include_router(terminal_binding_router)
|
||
|
||
async def _override_db():
|
||
yield db_session
|
||
|
||
admin_user = UserInfo(
|
||
employee_id="admin001",
|
||
name="管理员",
|
||
department="IT部",
|
||
avatar="",
|
||
roles=["admin"],
|
||
current_role="admin",
|
||
login_source="test",
|
||
)
|
||
|
||
app.dependency_overrides[get_db] = _override_db
|
||
app.dependency_overrides[get_current_user] = lambda: admin_user
|
||
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
yield ac
|
||
|
||
app.dependency_overrides.clear()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_binding(self, admin_client, db_session):
|
||
"""POST /itportal/admin/terminal-bindings — 新增绑定"""
|
||
resp = await admin_client.post("/itportal/admin/terminal-bindings", json={
|
||
"terminal_sn": "SN_TEST_001",
|
||
"terminal_name": "18F东区大屏",
|
||
"meetingroom_id": 1,
|
||
"meetingroom_name": "18F东区会议室",
|
||
"location": "18F东区",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert "id" in data["data"]
|
||
binding_id = data["data"]["id"]
|
||
|
||
# 验证数据已写入
|
||
from sqlalchemy import select
|
||
result = await db_session.execute(
|
||
select(TerminalRoomBinding).where(TerminalRoomBinding.id == binding_id)
|
||
)
|
||
binding = result.scalar_one_or_none()
|
||
assert binding is not None
|
||
assert binding.terminal_sn == "SN_TEST_001"
|
||
assert binding.meetingroom_id == 1
|
||
assert binding.is_active is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_binding_duplicate(self, admin_client, db_session):
|
||
"""重复绑定同一terminal_sn应报错"""
|
||
# 先创建一条
|
||
db_session.add(TerminalRoomBinding(
|
||
terminal_sn="SN_DUP_001",
|
||
terminal_name="终端1",
|
||
meetingroom_id=1,
|
||
meetingroom_name="会议室A",
|
||
location="18F",
|
||
is_active=True,
|
||
))
|
||
await db_session.flush()
|
||
|
||
# 再创建同SN的
|
||
resp = await admin_client.post("/itportal/admin/terminal-bindings", json={
|
||
"terminal_sn": "SN_DUP_001",
|
||
"terminal_name": "终端2",
|
||
"meetingroom_id": 2,
|
||
"meetingroom_name": "会议室B",
|
||
"location": "19F",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] != 0 # 应返回错误
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_bindings(self, admin_client, db_session):
|
||
"""GET /itportal/admin/terminal-bindings — 列表查询"""
|
||
# 插入测试数据
|
||
for i in range(3):
|
||
db_session.add(TerminalRoomBinding(
|
||
terminal_sn=f"SN_LIST_{i:03d}",
|
||
terminal_name=f"终端{i}",
|
||
meetingroom_id=i + 1,
|
||
meetingroom_name=f"会议室{i+1}",
|
||
location=f"{i+10}F",
|
||
is_active=True,
|
||
))
|
||
await db_session.flush()
|
||
|
||
resp = await admin_client.get("/itportal/admin/terminal-bindings")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
assert data["data"]["total"] >= 3
|
||
assert len(data["data"]["list"]) >= 3
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_bindings_with_keyword(self, admin_client, db_session):
|
||
"""GET /itportal/admin/terminal-bindings?keyword=xxx — 关键词搜索"""
|
||
db_session.add(TerminalRoomBinding(
|
||
terminal_sn="SN_KW_001",
|
||
terminal_name="大屏A",
|
||
meetingroom_id=1,
|
||
meetingroom_name="会议室A",
|
||
location="18F",
|
||
is_active=True,
|
||
))
|
||
db_session.add(TerminalRoomBinding(
|
||
terminal_sn="SN_KW_002",
|
||
terminal_name="大屏B",
|
||
meetingroom_id=2,
|
||
meetingroom_name="会议室B",
|
||
location="19F",
|
||
is_active=True,
|
||
))
|
||
await db_session.flush()
|
||
|
||
resp = await admin_client.get("/itportal/admin/terminal-bindings?keyword=大屏A")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
# 至少包含"大屏A"的记录
|
||
names = [item["terminal_name"] for item in data["data"]["list"]]
|
||
assert "大屏A" in names
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_binding(self, admin_client, db_session):
|
||
"""PUT /itportal/admin/terminal-bindings/{id} — 更新绑定"""
|
||
binding = TerminalRoomBinding(
|
||
terminal_sn="SN_UPD_001",
|
||
terminal_name="旧名称",
|
||
meetingroom_id=1,
|
||
meetingroom_name="旧会议室",
|
||
location="18F",
|
||
is_active=True,
|
||
)
|
||
db_session.add(binding)
|
||
await db_session.flush()
|
||
|
||
resp = await admin_client.put(f"/itportal/admin/terminal-bindings/{binding.id}", json={
|
||
"terminal_name": "新名称",
|
||
"meetingroom_name": "新会议室",
|
||
"is_active": False,
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
|
||
# 验证更新生效（flush 确保修改写入 session，再检查属性）
|
||
await db_session.flush()
|
||
assert binding.terminal_name == "新名称"
|
||
assert binding.meetingroom_name == "新会议室"
|
||
assert binding.is_active is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_binding_not_found(self, admin_client):
|
||
"""更新不存在的绑定应报错"""
|
||
resp = await admin_client.put("/itportal/admin/terminal-bindings/99999", json={
|
||
"terminal_name": "测试",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] != 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_binding(self, admin_client, db_session):
|
||
"""DELETE /itportal/admin/terminal-bindings/{id} — 删除绑定"""
|
||
binding = TerminalRoomBinding(
|
||
terminal_sn="SN_DEL_001",
|
||
terminal_name="待删除",
|
||
meetingroom_id=1,
|
||
meetingroom_name="会议室",
|
||
location="18F",
|
||
is_active=True,
|
||
)
|
||
db_session.add(binding)
|
||
await db_session.flush()
|
||
binding_id = binding.id
|
||
|
||
resp = await admin_client.delete(f"/itportal/admin/terminal-bindings/{binding_id}")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_binding_not_found(self, admin_client):
|
||
"""删除不存在的绑定应报错"""
|
||
resp = await admin_client.delete("/itportal/admin/terminal-bindings/99999")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["code"] != 0
|
||
|
||
|
||
# =============================================================================
|
||
# 模型测试
|
||
# =============================================================================
|
||
|
||
class TestModels:
|
||
"""模型字段定义测试"""
|
||
|
||
def test_terminal_room_binding_model(self):
|
||
"""TerminalRoomBinding 模型字段定义正确"""
|
||
# 验证表名
|
||
assert TerminalRoomBinding.__tablename__ == "terminal_room_binding"
|
||
|
||
# 验证字段存在
|
||
columns = {c.name for c in TerminalRoomBinding.__table__.columns}
|
||
expected = {
|
||
"id", "terminal_sn", "terminal_name", "meetingroom_id",
|
||
"meetingroom_name", "location", "is_active",
|
||
"created_at", "updated_at",
|
||
}
|
||
assert expected.issubset(columns), f"缺失字段: {expected - columns}"
|
||
|
||
# 验证 terminal_sn 唯一索引
|
||
sn_col = TerminalRoomBinding.__table__.columns["terminal_sn"]
|
||
assert sn_col.unique is True
|
||
|
||
# 验证 meetingroom_id 有索引（非唯一）
|
||
room_col = TerminalRoomBinding.__table__.columns["meetingroom_id"]
|
||
assert room_col.index is True
|
||
|
||
# 验证默认值
|
||
assert TerminalRoomBinding.__table__.columns["is_active"].default.arg is True
|
||
|
||
def test_meetingroom_booking_snapshot_model(self):
|
||
"""MeetingroomBookingSnapshot 模型字段定义正确"""
|
||
# 验证表名
|
||
assert MeetingroomBookingSnapshot.__tablename__ == "meetingroom_booking_snapshot"
|
||
|
||
# 验证字段存在
|
||
columns = {c.name for c in MeetingroomBookingSnapshot.__table__.columns}
|
||
expected = {
|
||
"id", "meetingroom_id", "booking_id", "subject", "booker",
|
||
"start_time", "end_time", "status", "snapshot_date", "created_at",
|
||
}
|
||
assert expected.issubset(columns), f"缺失字段: {expected - columns}"
|
||
|
||
# 验证索引字段
|
||
assert MeetingroomBookingSnapshot.__table__.columns["meetingroom_id"].index is True
|
||
assert MeetingroomBookingSnapshot.__table__.columns["booking_id"].index is True
|
||
assert MeetingroomBookingSnapshot.__table__.columns["snapshot_date"].index is True
|
||
|
||
# 验证默认值
|
||
assert MeetingroomBookingSnapshot.__table__.columns["status"].default.arg == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_terminal_room_binding_repr(self):
|
||
"""TerminalRoomBinding __repr__ 方法"""
|
||
binding = TerminalRoomBinding(
|
||
terminal_sn="SN_REPR_001",
|
||
terminal_name="测试终端",
|
||
meetingroom_id=1,
|
||
meetingroom_name="测试会议室",
|
||
)
|
||
repr_str = repr(binding)
|
||
assert "SN_REPR_001" in repr_str
|
||
assert "测试会议室" in repr_str
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_meetingroom_booking_snapshot_repr(self):
|
||
"""MeetingroomBookingSnapshot __repr__ 方法"""
|
||
from datetime import date
|
||
snapshot = MeetingroomBookingSnapshot(
|
||
meetingroom_id=1,
|
||
booking_id="bk_repr_001",
|
||
subject="测试快照",
|
||
start_time=datetime(2026, 7, 15, 10, 0, 0),
|
||
end_time=datetime(2026, 7, 15, 11, 0, 0),
|
||
snapshot_date=date(2026, 7, 15),
|
||
)
|
||
repr_str = repr(snapshot)
|
||
assert "bk_repr_001" in repr_str
|
||
assert "测试快照" in repr_str
|
||
|
||
|
||
# =============================================================================
|
||
# Pydantic Schema 测试
|
||
# =============================================================================
|
||
|
||
class TestSchemas:
|
||
"""Pydantic 请求/响应模型测试"""
|
||
|
||
def test_book_request_valid(self):
|
||
"""BookRequest 验证合法请求"""
|
||
req = BookRequest(
|
||
meetingroom_id=1,
|
||
subject="测试预定",
|
||
start_time="2026-07-15T10:00:00+08:00",
|
||
end_time="2026-07-15T11:00:00+08:00",
|
||
booker="zhangsan",
|
||
)
|
||
assert req.meetingroom_id == 1
|
||
assert req.subject == "测试预定"
|
||
|
||
def test_book_request_empty_subject(self):
|
||
"""BookRequest 空主题应验证失败"""
|
||
with pytest.raises(Exception):
|
||
BookRequest(
|
||
meetingroom_id=1,
|
||
subject="",
|
||
start_time="2026-07-15T10:00:00+08:00",
|
||
end_time="2026-07-15T11:00:00+08:00",
|
||
booker="zhangsan",
|
||
)
|
||
|
||
def test_terminal_binding_create_valid(self):
|
||
"""TerminalBindingCreate 验证"""
|
||
req = TerminalBindingCreate(
|
||
terminal_sn="SN001",
|
||
terminal_name="终端1",
|
||
meetingroom_id=1,
|
||
meetingroom_name="会议室1",
|
||
location="18F",
|
||
)
|
||
assert req.terminal_sn == "SN001"
|
||
|
||
def test_terminal_binding_create_empty_sn(self):
|
||
"""空terminal_sn应验证失败"""
|
||
with pytest.raises(Exception):
|
||
TerminalBindingCreate(
|
||
terminal_sn="",
|
||
meetingroom_id=1,
|
||
)
|
||
|
||
def test_terminal_binding_update_partial(self):
|
||
"""TerminalBindingUpdate 部分更新"""
|
||
req = TerminalBindingUpdate(terminal_name="新名称")
|
||
assert req.terminal_name == "新名称"
|
||
assert req.meetingroom_id is None
|
||
assert req.is_active is None
|