151 lines
5.4 KiB
Python
151 lines
5.4 KiB
Python
"""搜索 / 热搜 / 标签视频（5.9 / 5.10 / 5.11）：LIKE 匹配、敏感词置空、hot/latest 排序、热搜计数"""
|
||
from datetime import datetime, timedelta
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.models.common import SensitiveWord
|
||
from app.models.user import User
|
||
from app.models.video import Tag, Video, VideoTag
|
||
from app.services import video_service
|
||
|
||
|
||
def _identity(user_id: int | None = 1):
|
||
return SimpleNamespace(user_id=user_id, device_id=None)
|
||
|
||
|
||
@pytest.fixture
|
||
async def author(db):
|
||
u = User(password_hash="x", register_type=1, nickname="作者")
|
||
db.add(u)
|
||
await db.commit()
|
||
return u
|
||
|
||
|
||
async def _video(db, user_id, title, *, views=0, likes=0, comments=0,
|
||
published_at=None, status=None):
|
||
v = Video(
|
||
user_id=user_id, title=title,
|
||
status=status if status is not None else video_service.STATUS_PUBLISHED,
|
||
view_count=views, like_count=likes, comment_count=comments,
|
||
published_at=published_at or datetime.now(),
|
||
)
|
||
db.add(v)
|
||
await db.commit()
|
||
return v
|
||
|
||
|
||
# ---------- 5.10 视频搜索 ----------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_by_title(db, redis, author):
|
||
await _video(db, author.id, "美食教程：红烧肉")
|
||
await _video(db, author.id, "旅游日记")
|
||
|
||
r = await video_service.search_videos(db, redis, _identity(), "美食", "hot", 1, 20)
|
||
assert r["total"] == 1
|
||
assert "美食" in r["list"][0]["title"]
|
||
# 搜索行为计入热搜 ZSet
|
||
assert redis.zsets[video_service.HOT_SEARCH_KEY]["美食"] == 1.0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_by_tag(db, redis, author):
|
||
v = await _video(db, author.id, "增肌训练记录")
|
||
tag = Tag(name="健身")
|
||
db.add(tag)
|
||
await db.flush()
|
||
db.add(VideoTag(video_id=v.id, tag_id=tag.id))
|
||
await db.commit()
|
||
|
||
r = await video_service.search_videos(db, redis, _identity(), "健身", "hot", 1, 20)
|
||
assert r["total"] == 1
|
||
assert r["list"][0]["video_id"] == v.id
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_sensitive_word_returns_empty(db, redis, author):
|
||
await _video(db, author.id, "包含违禁词的视频")
|
||
db.add(SensitiveWord(word="违禁"))
|
||
await db.commit()
|
||
|
||
r = await video_service.search_videos(db, redis, _identity(), "违禁", "hot", 1, 20)
|
||
assert r == {"total": 0, "list": []}
|
||
# 敏感词搜索不计入热搜
|
||
assert video_service.HOT_SEARCH_KEY not in redis.zsets
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_unpublished_not_matched(db, redis, author):
|
||
await _video(db, author.id, "待审核美食", status=1)
|
||
r = await video_service.search_videos(db, redis, _identity(), "美食", "hot", 1, 20)
|
||
assert r["total"] == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_hot_and_latest_sort(db, redis, author):
|
||
# hot 分 = view×1 + like×5 + comment×10
|
||
low = await _video(db, author.id, "测试甲", views=10, likes=0, comments=0) # 10
|
||
high = await _video(db, author.id, "测试乙", views=0, likes=3, comments=0) # 15
|
||
|
||
r = await video_service.search_videos(db, redis, _identity(), "测试", "hot", 1, 20)
|
||
assert [i["video_id"] for i in r["list"]] == [high.id, low.id]
|
||
|
||
old = await _video(db, author.id, "测试旧", published_at=datetime.now() - timedelta(days=2))
|
||
r = await video_service.search_videos(db, redis, _identity(), "测试", "latest", 1, 20)
|
||
assert r["list"][0]["video_id"] in {high.id, low.id} # 最新发布的在前
|
||
assert r["list"][-1]["video_id"] == old.id
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_pagination(db, redis, author):
|
||
for i in range(5):
|
||
await _video(db, author.id, f"分页视频{i}")
|
||
r1 = await video_service.search_videos(db, redis, _identity(), "分页", "latest", 1, 2)
|
||
r2 = await video_service.search_videos(db, redis, _identity(), "分页", "latest", 2, 2)
|
||
assert r1["total"] == 5 and len(r1["list"]) == 2 and len(r2["list"]) == 2
|
||
assert {i["video_id"] for i in r1["list"]}.isdisjoint({i["video_id"] for i in r2["list"]})
|
||
|
||
|
||
# ---------- 5.11 热搜榜 ----------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_hot_search_top10(db, redis):
|
||
for i in range(12):
|
||
await redis.zincrby(video_service.HOT_SEARCH_KEY, i + 1, f"词{i}")
|
||
r = await video_service.get_hot_search(redis)
|
||
assert len(r["words"]) == video_service.HOT_SEARCH_TOP
|
||
# 按计数倒序：词11 计数最高
|
||
assert r["words"][0] == "词11"
|
||
assert r["words"][-1] == "词2"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_hot_search_empty(redis):
|
||
assert (await video_service.get_hot_search(redis)) == {"words": []}
|
||
|
||
|
||
# ---------- 5.9 标签视频列表 ----------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tag_videos(db, redis, author):
|
||
tag = Tag(name="美食")
|
||
db.add(tag)
|
||
await db.flush()
|
||
v1 = await _video(db, author.id, "红烧肉")
|
||
v2 = await _video(db, author.id, "清蒸鱼")
|
||
v3 = await _video(db, author.id, "待审核的菜", status=1)
|
||
for v in (v1, v2, v3):
|
||
db.add(VideoTag(video_id=v.id, tag_id=tag.id))
|
||
await db.commit()
|
||
|
||
r = await video_service.get_tag_videos(db, _identity(), "美食", "hot", 1, 20)
|
||
assert r["tag"] == "美食" and r["total"] == 2 # 未发布不计
|
||
assert {i["video_id"] for i in r["list"]} == {v1.id, v2.id}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tag_videos_tag_not_exists(db):
|
||
r = await video_service.get_tag_videos(db, _identity(), "不存在", "hot", 1, 20)
|
||
assert r == {"tag": "不存在", "total": 0, "list": []}
|