Simon
279 lines
11 KiB
Python
279 lines
11 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — 头像同步服务单元测试
|
||
# =============================================================================
|
||
# 验证 #75「头像同步功能完善」的核心交付物：
|
||
# 1. avatar_service.sync_employee_avatar
|
||
# - 员工存在：更新 avatar + 刷新 avatar_updated_at + 删除 Redis 缓存
|
||
# - 空 avatar：不更新（保持容错，不覆盖已有头像 / 不清缓存）
|
||
# - 异常路径：内部吞掉，绝不向上抛出（不阻塞登录）
|
||
# - corp_id 过滤：只更新 (corp_id, employee_id) 匹配的员工
|
||
# 2. avatar_service.clean_avatar_url
|
||
# - 带 ? 查询参数的企微 URL 正确去参
|
||
# - 无参数 URL 原样返回；空/None 输入返回空串
|
||
# 3. session_service.SessionService.AVATAR_CACHE_TTL 已由 7 天变为 1 天
|
||
# 4. session_service._get_employee_avatar
|
||
# - 返回前 clean_avatar_url（去参）
|
||
# - 回源后回写稳定 URL 并以 1 天 TTL 缓存
|
||
#
|
||
# 运行：backend/venv/Scripts/python.exe -m pytest tests/test_avatar_service.py -v
|
||
# =============================================================================
|
||
|
||
import pytest
|
||
|
||
from app.config import settings
|
||
from app.models.employee import Employee
|
||
from app.services.avatar_service import clean_avatar_url, sync_employee_avatar
|
||
from app.services.session_service import SessionService
|
||
from sqlalchemy import select
|
||
from tests.conftest import MockRedis
|
||
|
||
|
||
# =============================================================================
|
||
# clean_avatar_url 单测
|
||
# =============================================================================
|
||
class TestCleanAvatarUrl:
|
||
"""清理企微头像 URL 的查询参数。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strips_query_string(self):
|
||
"""带 ? 查询参数的企微 URL 应去掉参数，保留稳定部分。"""
|
||
url = "https://wework.qpic.cn/wwpic/abc123.png?size=96&t=1718000000"
|
||
assert clean_avatar_url(url) == "https://wework.qpic.cn/wwpic/abc123.png"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_url_without_query_unchanged(self):
|
||
"""无查询参数的 URL 应原样返回。"""
|
||
url = "https://wework.qpic.cn/wwpic/abc123.png"
|
||
assert clean_avatar_url(url) == url
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_only_query_marker_stripped(self):
|
||
"""只有 ? 而无参数时，应去掉 ? 及之后（这里之后为空）。"""
|
||
url = "https://wework.qpic.cn/wwpic/abc123.png?"
|
||
assert clean_avatar_url(url) == "https://wework.qpic.cn/wwpic/abc123.png"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_string_returns_empty(self):
|
||
"""空字符串输入返回空串。"""
|
||
assert clean_avatar_url("") == ""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_none_returns_empty(self):
|
||
"""None 输入返回空串（不抛异常，与 sync 的容错一致）。"""
|
||
assert clean_avatar_url(None) == ""
|
||
|
||
|
||
# =============================================================================
|
||
# sync_employee_avatar 单测
|
||
# =============================================================================
|
||
class TestSyncEmployeeAvatar:
|
||
"""统一「写库 avatar + 刷新 avatar_updated_at + 删 Redis 缓存」。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_updates_avatar_and_clears_cache_when_employee_exists(
|
||
self, db_session, mock_redis
|
||
):
|
||
"""员工存在时：avatar 更新、avatar_updated_at 刷新、Redis 缓存被删除。"""
|
||
emp = Employee(
|
||
employee_id="emp_sync_001",
|
||
corp_id=settings.wecom_corp_id,
|
||
name="同步测试员工",
|
||
avatar="",
|
||
)
|
||
db_session.add(emp)
|
||
await db_session.flush()
|
||
|
||
# 预置旧缓存，验证 sync 会把它删掉
|
||
cache_key = f"employee:avatar:emp_sync_001"
|
||
await mock_redis.set(cache_key, "https://old.example/old.png")
|
||
|
||
raw_avatar = "https://wework.qpic.cn/wwpic/new.png?size=96&t=1"
|
||
await sync_employee_avatar(db_session, mock_redis, "emp_sync_001", raw_avatar)
|
||
|
||
# 重新查询，确认已落库
|
||
result = await db_session.execute(
|
||
select(Employee).where(
|
||
Employee.employee_id == "emp_sync_001",
|
||
Employee.corp_id == settings.wecom_corp_id,
|
||
)
|
||
)
|
||
persisted = result.scalars().first()
|
||
assert persisted is not None
|
||
# avatar 应为去参后的稳定 URL
|
||
assert persisted.avatar == "https://wework.qpic.cn/wwpic/new.png"
|
||
assert "?" not in persisted.avatar
|
||
# 头像更新时间被刷新（非空）
|
||
assert persisted.avatar_updated_at is not None
|
||
# Redis 缓存被删除
|
||
assert await mock_redis.get(cache_key) is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_avatar_does_not_overwrite(self, db_session, mock_redis):
|
||
"""传入空 avatar 时不更新（保持容错），也不删缓存。"""
|
||
emp = Employee(
|
||
employee_id="emp_sync_002",
|
||
corp_id=settings.wecom_corp_id,
|
||
name="容错测试员工",
|
||
avatar="https://wework.qpic.cn/wwpic/existing.png",
|
||
)
|
||
db_session.add(emp)
|
||
await db_session.flush()
|
||
|
||
cache_key = f"employee:avatar:emp_sync_002"
|
||
await mock_redis.set(cache_key, "cached-old")
|
||
|
||
# 企微未返回头像（空串）
|
||
await sync_employee_avatar(db_session, mock_redis, "emp_sync_002", "")
|
||
|
||
result = await db_session.execute(
|
||
select(Employee).where(
|
||
Employee.employee_id == "emp_sync_002",
|
||
Employee.corp_id == settings.wecom_corp_id,
|
||
)
|
||
)
|
||
persisted = result.scalars().first()
|
||
# 已有头像不应被清空
|
||
assert persisted.avatar == "https://wework.qpic.cn/wwpic/existing.png"
|
||
# 未刷新更新时间
|
||
assert persisted.avatar_updated_at is None
|
||
# 缓存不应被删（early return，没走到 delete）
|
||
assert await mock_redis.get(cache_key) is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_exception_does_not_propagate(self, mock_redis):
|
||
"""DB 查询异常时内部吞掉，函数不抛出（不阻塞登录）。"""
|
||
# 一个会在 execute 时抛异常的假 db
|
||
class BoomDb:
|
||
async def execute(self, *args, **kwargs):
|
||
raise RuntimeError("模拟数据库故障")
|
||
|
||
async def commit(self, *args, **kwargs):
|
||
pass
|
||
|
||
# 不应抛出
|
||
await sync_employee_avatar(
|
||
BoomDb(),
|
||
mock_redis,
|
||
"emp_sync_boom",
|
||
"https://wework.qpic.cn/wwpic/x.png?a=1",
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_redis_delete_error_does_not_propagate(self, db_session):
|
||
"""Redis 删除失败时内部吞掉，函数不抛出。"""
|
||
emp = Employee(
|
||
employee_id="emp_sync_003",
|
||
corp_id=settings.wecom_corp_id,
|
||
name="Redis异常测试",
|
||
avatar="",
|
||
)
|
||
db_session.add(emp)
|
||
await db_session.flush()
|
||
|
||
# delete 会抛异常的假 redis
|
||
class BoomRedis:
|
||
async def delete(self, *names):
|
||
raise RuntimeError("模拟Redis故障")
|
||
|
||
# 不应抛出
|
||
await sync_employee_avatar(
|
||
db_session,
|
||
BoomRedis(),
|
||
"emp_sync_003",
|
||
"https://wework.qpic.cn/wwpic/y.png?a=1",
|
||
)
|
||
|
||
# 仍应完成写库（avatar 被更新）
|
||
result = await db_session.execute(
|
||
select(Employee).where(
|
||
Employee.employee_id == "emp_sync_003",
|
||
Employee.corp_id == settings.wecom_corp_id,
|
||
)
|
||
)
|
||
persisted = result.scalars().first()
|
||
assert persisted.avatar == "https://wework.qpic.cn/wwpic/y.png"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_only_matches_same_corp_id(self, db_session, mock_redis):
|
||
"""corp_id 不匹配的员工不应被更新（复合唯一键正确性）。"""
|
||
target_id = "emp_sync_004"
|
||
# 同 employee_id，但 corp_id 不同（模拟上下游互联企业）
|
||
other_corp_emp = Employee(
|
||
employee_id=target_id,
|
||
corp_id="another_corp_id",
|
||
name="其他企业员工",
|
||
avatar="https://wework.qpic.cn/wwpic/other.png",
|
||
)
|
||
db_session.add(other_corp_emp)
|
||
await db_session.flush()
|
||
|
||
cache_key = f"employee:avatar:{target_id}"
|
||
await mock_redis.set(cache_key, "cached")
|
||
|
||
# 用当前 corp_id 去同步 —— 应只命中当前企业的员工（这里不存在）
|
||
await sync_employee_avatar(
|
||
db_session,
|
||
mock_redis,
|
||
target_id,
|
||
"https://wework.qpic.cn/wwpic/new.png?a=1",
|
||
)
|
||
|
||
result = await db_session.execute(
|
||
select(Employee).where(Employee.employee_id == target_id)
|
||
)
|
||
rows = result.scalars().all()
|
||
# sync_employee_avatar 只更新已存在员工、不创建新记录：
|
||
# 当前 corp_id 下无该 employee_id，因此仍只有另一条企业的 1 条记录
|
||
assert len(rows) == 1
|
||
other = rows[0]
|
||
assert other.corp_id == "another_corp_id"
|
||
# 其他企业的员工头像未被覆盖（corp_id 过滤生效）
|
||
assert other.avatar == "https://wework.qpic.cn/wwpic/other.png"
|
||
assert other.avatar_updated_at is None
|
||
# 但缓存仍被删除（delete 不依赖 employee 是否存在）
|
||
assert await mock_redis.get(cache_key) is None
|
||
|
||
|
||
# =============================================================================
|
||
# session_service 头像缓存 TTL / 清理 单测
|
||
# =============================================================================
|
||
class TestSessionServiceAvatar:
|
||
"""会话服务的头像缓存 TTL 与 clean_avatar_url 集成。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_avatar_cache_ttl_is_one_day(self):
|
||
"""头像缓存 TTL 已由 7 天改为 1 天（要求 B）。"""
|
||
assert SessionService.AVATAR_CACHE_TTL == 1 * 24 * 60 * 60
|
||
assert SessionService.AVATAR_CACHE_TTL != 7 * 24 * 60 * 60
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_employee_avatar_cleans_and_caches_one_day(
|
||
self, db_session, mock_redis
|
||
):
|
||
"""_get_employee_avatar：返回去参后的稳定 URL，并以 1 天 TTL 回写缓存。"""
|
||
emp_id = "emp_sess_001"
|
||
Employee.__table__ # 确保已注册
|
||
emp = Employee(
|
||
employee_id=emp_id,
|
||
corp_id=settings.wecom_corp_id,
|
||
name="缓存测试员工",
|
||
# DB 里存的是带参数的 URL
|
||
avatar="https://wework.qpic.cn/wwpic/db.png?size=96&t=9",
|
||
)
|
||
db_session.add(emp)
|
||
await db_session.flush()
|
||
|
||
svc = SessionService(db_session, wecom_service=None, redis_client=mock_redis)
|
||
got = await svc._get_employee_avatar(emp_id)
|
||
|
||
# 返回的是去参后的稳定 URL
|
||
assert got == "https://wework.qpic.cn/wwpic/db.png"
|
||
assert "?" not in got
|
||
|
||
# Redis 中已缓存，且 TTL 为 1 天
|
||
cache_key = f"employee:avatar:{emp_id}"
|
||
cached = await mock_redis.get(cache_key)
|
||
assert cached is not None
|
||
assert cached.decode("utf-8") == "https://wework.qpic.cn/wwpic/db.png"
|
||
assert mock_redis._ttl.get(cache_key) == 1 * 24 * 60 * 60
|