fix: restore tmdb trending recommendations

This commit is contained in:
jxxghp
2026-05-20 10:55:01 +08:00
parent c3e4e1a764
commit a5745af484
6 changed files with 112 additions and 20 deletions

View File

@@ -107,6 +107,22 @@ class MediaRecognizeModulesTest(TestCase):
self.assertEqual(result, "zh,en,null,ja")
def test_tmdb_trending_filters_non_media_and_normalizes_media_type(self):
"""TMDB流行趋势应过滤人物项并把字符串媒体类型转为内部枚举。"""
infos = [
{"id": 100, "media_type": "movie", "title": "测试电影"},
{"id": 101, "media_type": "tv", "name": "测试剧集"},
{"id": 102, "media_type": "person", "name": "测试人物"},
{"id": 103, "media_type": MediaType.MOVIE, "title": "枚举电影"},
]
result = TmdbApi._normalize_trending_infos(infos)
self.assertEqual([info["id"] for info in result], [100, 101, 103])
self.assertEqual(result[0]["media_type"], MediaType.MOVIE)
self.assertEqual(result[1]["media_type"], MediaType.TV)
self.assertEqual(result[2]["media_type"], MediaType.MOVIE)
def test_tmdb_obtain_images_uses_language_fallback_and_picks_best(self):
"""obtain_images 应从图片接口回填缺失的海报和背景图。"""
module = TheMovieDbModule()

View File

@@ -0,0 +1,42 @@
import asyncio
from unittest import TestCase
from unittest.mock import AsyncMock, patch
from app.chain.recommend import RecommendChain
from app.core.cache import TTLCache
class RecommendChainTest(TestCase):
def tearDown(self):
"""
清理推荐缓存,避免缓存装饰器状态影响其他用例。
"""
RecommendChain.tmdb_trending.cache_clear()
asyncio.run(RecommendChain.async_tmdb_trending.cache_clear())
TTLCache(region=RecommendChain.recommend_cache_region).clear()
def test_tmdb_trending_does_not_cache_empty_result(self):
"""
TMDB流行趋势返回空列表时不应缓存避免一次接口异常后长时间固定为空。
"""
chain = RecommendChain()
with patch("app.chain.recommend.TmdbChain") as tmdb_chain:
tmdb_chain.return_value.tmdb_trending.side_effect = [[], []]
self.assertEqual(chain.tmdb_trending(page=1), [])
self.assertEqual(chain.tmdb_trending(page=1), [])
self.assertEqual(tmdb_chain.return_value.tmdb_trending.call_count, 2)
def test_async_tmdb_trending_does_not_cache_empty_result(self):
"""
异步TMDB流行趋势返回空列表时也不应缓存。
"""
chain = RecommendChain()
with patch("app.chain.recommend.TmdbChain") as tmdb_chain:
tmdb_chain.return_value.async_run_module = AsyncMock(side_effect=[[], []])
self.assertEqual(asyncio.run(chain.async_tmdb_trending(page=1)), [])
self.assertEqual(asyncio.run(chain.async_tmdb_trending(page=1)), [])
self.assertEqual(tmdb_chain.return_value.async_run_module.call_count, 2)

View File

@@ -146,13 +146,13 @@ class _UnicodeDecodeErrorResponse:
模拟 httpx.Response.json() 直接抛 UnicodeDecodeError 的异常响应。
"""
def __init__(self, content: bytes = b"\x8b"):
def __init__(self, content: bytes = b"\x8b", text: str = ""):
"""
初始化一个带有压缩响应特征的伪响应对象。
"""
self.headers = {"Content-Type": "application/json", "Content-Encoding": "gzip"}
self.status_code = 200
self.text = ""
self.text = text
self.content = content
def json(self):
@@ -227,10 +227,16 @@ class TmdbResponseCacheTest(TestCase):
错误编码的响应体也应转换为TMDbException避免UnicodeDecodeError直接冒泡。
"""
tmdb = TMDb()
tmdb._req.get_res = lambda *args, **kwargs: _UnicodeDecodeErrorResponse()
tmdb._req.get_res = lambda *args, **kwargs: _UnicodeDecodeErrorResponse(
text="乱码内容不应进入日志"
)
with self.assertRaisesRegex(TMDbException, "不是有效JSON.*Content-Encodinggzip"):
with self.assertRaisesRegex(
TMDbException,
"不是有效JSON.*Content-Encodinggzip.*响应内容因编码错误已省略",
) as cm:
TMDb.request.__wrapped__(tmdb, "GET", "https://example.com", None, None)
self.assertNotIn("乱码内容", str(cm.exception))
def test_request_decodes_raw_gzip_json_response(self):
"""