mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
123 lines
5.6 KiB
Python
123 lines
5.6 KiB
Python
"""
|
|
验证通知独立海报经过渲染、持久化、查询和事件投影，且不改变渠道图片。
|
|
"""
|
|
|
|
import asyncio
|
|
import importlib
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
from app.api.endpoints.message import get_notification_message
|
|
from app.application.messaging.message import MessageQueryService, MessageTemplateHelper
|
|
from app.chain.base import ChainBase
|
|
from app.db.oper.message import MessageOper
|
|
from app.db.session import AsyncSessionFactory
|
|
from app.domain.context import MediaInfo, MusicInfo
|
|
from app.schemas.message import Message
|
|
|
|
|
|
@pytest.mark.parametrize("has_poster", [True, False])
|
|
def test_render_preserves_backdrop_and_projects_only_real_poster(has_poster: bool) -> None:
|
|
"""
|
|
即使消息已经有正文也应提取独立海报，缺失时不得从背景图回退。
|
|
"""
|
|
media = MediaInfo(
|
|
title="测试影片",
|
|
backdrop_path="https://images.example.com/backdrop.jpg",
|
|
poster_path="https://images.example.com/poster.jpg" if has_poster else None,
|
|
)
|
|
message = Message(title="已完成整理", image=media.get_message_image())
|
|
|
|
rendered = MessageTemplateHelper.render(message, mediainfo=media)
|
|
|
|
assert rendered is message
|
|
assert message.image == "https://images.example.com/backdrop.jpg"
|
|
assert message.poster == ("https://images.example.com/poster.jpg" if has_poster else None)
|
|
event = ChainBase._build_notice_message_data(message)
|
|
assert event["poster"] == message.poster
|
|
assert event["image"] == message.image
|
|
|
|
|
|
def test_render_keeps_explicit_poster_and_supports_music_cover() -> None:
|
|
"""
|
|
显式海报优先，音乐媒体使用已有封面，无媒体上下文不猜测图片类型。
|
|
"""
|
|
music = MusicInfo(title="测试专辑", cover_url="https://images.example.com/album.jpg")
|
|
message = Message(title="入库完成", poster="https://images.example.com/custom.jpg")
|
|
assert MessageTemplateHelper.render(message, mediainfo=music).poster == message.poster
|
|
assert MessageTemplateHelper.render(Message(title="专辑通知"), mediainfo=music).poster == music.cover_url
|
|
assert MessageTemplateHelper.render(Message(title="普通通知", image=music.cover_url)).poster is None
|
|
|
|
|
|
@pytest.mark.parametrize("async_write", [False, True])
|
|
def test_notification_history_preserves_poster_and_legacy_null(async_write: bool) -> None:
|
|
"""
|
|
同步和异步写入均应通过真实通知查询返回独立海报，同时兼容旧消息。
|
|
"""
|
|
title = f"poster-contract-{uuid4().hex}"
|
|
payload = {
|
|
"title": title,
|
|
"image": "https://images.example.com/backdrop.jpg",
|
|
"poster": "https://images.example.com/poster.jpg",
|
|
}
|
|
|
|
async def write_and_query() -> None:
|
|
"""
|
|
仅使用测试临时数据库保存消息，并经 API schema 读取结果。
|
|
"""
|
|
async with AsyncSessionFactory() as db:
|
|
if async_write:
|
|
await MessageOper(db).async_add(**payload)
|
|
await MessageOper(db).async_add(title=f"{title}-legacy", image=payload["image"])
|
|
await db.commit()
|
|
else:
|
|
MessageOper().add(**payload)
|
|
MessageOper().add(title=f"{title}-legacy", image=payload["image"])
|
|
messages = await get_notification_message(service=MessageQueryService(MessageOper(db)))
|
|
current = next(item for item in messages if item.title == title)
|
|
legacy = next(item for item in messages if item.title == f"{title}-legacy")
|
|
assert current.image == payload["image"]
|
|
assert current.poster == payload["poster"]
|
|
assert legacy.image == payload["image"]
|
|
assert legacy.poster is None
|
|
|
|
asyncio.run(write_and_query())
|
|
|
|
|
|
@pytest.mark.parametrize("already_created", [False, True])
|
|
def test_message_poster_migration_preserves_existing_images(monkeypatch, already_created: bool) -> None:
|
|
"""
|
|
迁移支持旧表和新 ORM 已建表，重复执行及回滚不会改写历史背景图。
|
|
"""
|
|
migration = importlib.import_module("database.versions.e4a8c7b29016_3_1_9")
|
|
engine = sa.create_engine("sqlite://")
|
|
metadata = sa.MetaData()
|
|
columns = [sa.Column("id", sa.Integer(), primary_key=True), sa.Column("image", sa.String())]
|
|
if already_created:
|
|
columns.append(sa.Column("poster", sa.String(), nullable=True))
|
|
messages = sa.Table("message", metadata, *columns)
|
|
with engine.begin() as connection:
|
|
metadata.create_all(connection)
|
|
connection.execute(messages.insert(), {"id": 1, "image": "legacy-backdrop.jpg"})
|
|
monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection)))
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
migrated = sa.Table("message", sa.MetaData(), autoload_with=connection)
|
|
row = connection.execute(sa.select(migrated)).mappings().one()
|
|
assert row["image"] == "legacy-backdrop.jpg"
|
|
assert row["poster"] is None
|
|
connection.execute(migrated.insert(), {"id": 2, "image": "backdrop.jpg", "poster": "poster.jpg"})
|
|
assert connection.execute(sa.select(migrated.c.poster).where(migrated.c.id == 2)).scalar_one() == "poster.jpg"
|
|
migration.downgrade()
|
|
migration.downgrade()
|
|
restored = sa.Table("message", sa.MetaData(), autoload_with=connection)
|
|
assert "poster" not in restored.c
|
|
assert connection.execute(sa.select(restored.c.image).order_by(restored.c.id)).scalars().all() == [
|
|
"legacy-backdrop.jpg", "backdrop.jpg",
|
|
]
|
|
engine.dispose()
|