MoviePilot/tests/test_site_repository.py

"""
站点 SQLAlchemy 适配器的快照与事务边界测试。
"""

import pytest

from app.application.security.image import SiteImageDomainCache
from app.application.site.contract import (
    SiteMutation,
    SitePriorityMutation,
    SiteSnapshot,
    SiteStatisticSnapshot,
    SiteUserDataMutation,
    SiteUserDataSnapshot,
)
from app.application.site.query import SiteQueryService
from app.db.adapters import site as site_adapter
from app.db.adapters.site import SessionSiteRepository, TransactionalSiteRepository
from app.db.models.site import Site
from app.db.models.siteicon import SiteIcon
from app.db.models.sitestatistic import SiteStatistic
from app.db.models.siteuserdata import SiteUserData
from app.db.session import SessionFactory, async_session_scope
from app.db.uow import SqlAlchemyAsyncUnitOfWork


def _transactional_repository() -> TransactionalSiteRepository:
    """
    按生产组合方式构造短 Session 站点仓储。
    """
    return TransactionalSiteRepository(
        sync_session=SessionFactory,
        async_session=async_session_scope,
    )


@pytest.mark.asyncio
async def test_site_writes_refresh_image_domains(db, monkeypatch) -> None:
    """
    同步新增、同步改址和异步改址提交后均撤销旧图片域名快照。
    """
    db.watermark(Site)
    cache = SiteImageDomainCache()
    monkeypatch.setattr(site_adapter, "site_image_domains", cache)
    repository = _transactional_repository()
    source = SiteQueryService(repository=repository)
    original = await cache.get(source)
    assert repository.add(SiteMutation({
        "name": "图片缓存", "domain": "image-cache.example", "url": "https://old-image.example/",
    })).success
    snapshot = repository.get_by_domain("image-cache.example")
    assert snapshot is not None
    assert "https://old-image.example" in await cache.get(source) - original
    repository.update(snapshot.id, SiteMutation({"url": "https://new-image.example/"}))
    domains = await cache.get(source)
    assert "https://old-image.example" not in domains
    assert "https://new-image.example" in domains
    await repository.async_update(snapshot.id, SiteMutation({"url": "https://async-image.example/"}))
    domains = await cache.get(source)
    assert "https://new-image.example" not in domains
    assert "https://async-image.example" in domains


def test_transactional_repository_projects_frozen_site_snapshot(db) -> None:
    """
    独立仓储应在 Session 内复制 JSON 并返回不可变站点快照。
    """
    db.watermark(Site)
    note = {"nested": ["before"]}
    repository = _transactional_repository()

    result = repository.add(
        SiteMutation(
            {
                "name": "快照站点",
                "domain": "typed-site.example",
                "url": "https://typed-site.example/",
                "note": note,
                "is_active": True,
            }
        )
    )
    note["nested"].append("after")
    snapshot = repository.get_by_domain("typed-site.example")

    assert result.success is True
    assert isinstance(snapshot, SiteSnapshot)
    assert snapshot.note == {"nested": ["before"]}
    with pytest.raises(TypeError, match="不可修改"):
        # 主动修改快照中的嵌套数据，验证对外投影与持久对象隔离。
        snapshot.note["new"] = True  # type: ignore[index]


@pytest.mark.asyncio
async def test_transactional_repository_projects_related_snapshots(db) -> None:
    """
    用户数据、图标与统计查询均不得把 ORM 对象带出 Session。
    """
    db.add(
        SiteUserData(
            domain="related-site.example",
            name="关联站点",
            seeding_info={"size": [1, 2]},
            message_unread_contents=[{"title": "notice"}],
            err_msg="",
        ),
        SiteIcon(
            name="关联站点",
            domain="related-site.example",
            url="https://related-site.example/favicon.ico",
        ),
        SiteStatistic(
            domain="related-site.example",
            success=2,
            fail=1,
            note={"2026-08-28 10:00:00": 3},
        ),
    )
    repository = _transactional_repository()

    userdata = await repository.async_get_userdata_by_domain("related-site.example")
    icon = await repository.async_get_icon_by_domain("related-site.example")
    statistic = await repository.async_get_statistic_by_domain("related-site.example")

    assert isinstance(userdata[0], SiteUserDataSnapshot)
    assert userdata[0].seeding_info == {"size": [1, 2]}
    assert icon is not None and icon.domain == "related-site.example"
    assert isinstance(statistic, SiteStatisticSnapshot)
    assert statistic.note == {"2026-08-28 10:00:00": 3}


@pytest.mark.parametrize(
    ("ratio_fields", "expected_ratio"),
    [({}, 0), ({"ratio": None}, None), ({"ratio": 0}, 0), ({"ratio": 1.23}, 1.23)],
)
def test_userdata_ratio_preserves_null_and_default_through_storage(db, ratio_fields, expected_ratio) -> None:
    """
    首次写入与当天更新均保留显式空分享率，省略字段仍使用默认零值。
    """
    db.watermark(SiteUserData)
    domain = "nullable-ratio.example"
    repository = _transactional_repository()
    assert repository.update_userdata(
        domain, "空分享率站点", SiteUserDataMutation({"download": 0, **ratio_fields}),
    ).success

    def read_ratio():
        """
        通过生产查询 DTO 读取已提交的分享率。
        """
        records = SiteQueryService(repository).userdata_latest_sync()
        return next(item.ratio for item in records if item.domain == domain)

    assert read_ratio() == expected_ratio
    repository.update_userdata(domain, "空分享率站点", SiteUserDataMutation({"ratio": 2.5}))
    assert read_ratio() == 2.5
    repository.update_userdata(domain, "空分享率站点", SiteUserDataMutation({"ratio": None}))
    assert read_ratio() is None


def test_userdata_default_seeding_info_is_compatible_with_sync_query_dto(db) -> None:
    """
    省略做种明细时，查询 DTO 应收到空列表而不是 JSON 对象。
    """
    db.watermark(SiteUserData)
    domain = "default-seeding-info.example"
    db.add(SiteUserData(domain=domain, name="默认做种明细站点", err_msg=""))

    result = SiteQueryService(_transactional_repository()).userdata_latest_sync()

    userdata = next(item for item in result if item.domain == domain)
    assert userdata.seeding_info == []


@pytest.mark.asyncio
async def test_session_repository_leaves_commit_and_rollback_to_caller(db) -> None:
    """
    请求级仓储仅暂存，外层回滚和提交应分别决定最终可见性。
    """
    db.watermark(Site)
    repository = _transactional_repository()
    domain = "request-site.example"

    async with async_session_scope() as session:
        request_repository = SessionSiteRepository(session)
        await request_repository.stage_create(
            SiteMutation(
                {
                    "name": "请求站点",
                    "domain": domain,
                    "url": f"https://{domain}/",
                    "pri": 1,
                }
            )
        )
        assert await request_repository.async_get_by_domain(domain) is not None
        await session.rollback()

    assert repository.get_by_domain(domain) is None

    async with async_session_scope() as session:
        request_repository = SessionSiteRepository(session)
        await request_repository.stage_create(
            SiteMutation(
                {
                    "name": "请求站点",
                    "domain": domain,
                    "url": f"https://{domain}/",
                    "pri": 1,
                }
            )
        )
        await SqlAlchemyAsyncUnitOfWork(session).commit()

    snapshot = repository.get_by_domain(domain)
    assert snapshot is not None and snapshot.pri == 1

    async with async_session_scope() as session:
        request_repository = SessionSiteRepository(session)
        await request_repository.stage_priorities((SitePriorityMutation(site_id=snapshot.id, priority=8),))
        await SqlAlchemyAsyncUnitOfWork(session).commit()

    # 本用例已构造非空数据，直接检查或修改其内容以验证快照契约。
    assert repository.get(snapshot.id).pri == 8  # type: ignore[union-attr]