Files
MoviePilot/tests/test_site_query_service.py
jxxghp 6ca7b294a0 fix(download): 共享站点流控并保留频控错误原因 (#7246)
* fix(download): share site limits and preserve frequency errors

* test(startup): record shared site throttle module
2026-10-30 14:53:32 +08:00

62 lines
1.9 KiB
Python

"""
站点查询应用服务的稳定快照投影契约测试。
"""
from unittest.mock import Mock
import pytest
from app.application.site.contract import SiteSnapshot, SiteUserDataSnapshot
from app.application.site.query import SiteQueryService
from app.schemas.site import SiteUserData
@pytest.mark.parametrize("found", [False, True])
def test_sync_domain_query_projects_config_for_request_throttle(found) -> None:
"""
独立种子下载按规范域名读取脱离 Session 的流控配置,不引入异步查询。
"""
repository = Mock()
repository.get_by_domain.return_value = (
SiteSnapshot(id=42, name="站点", url="https://tracker.example/", limit_seconds=10,
limit_interval=60, limit_count=8) if found else None
)
result = SiteQueryService(repository).get_by_domain_sync("tracker.example")
repository.get_by_domain.assert_called_once_with("tracker.example")
if found:
assert (result.id, result.limit_seconds, result.limit_interval, result.limit_count) == (42, 10, 60, 8)
else:
assert result is None
def test_userdata_latest_sync_projects_snapshot_to_dto() -> None:
"""
同步站点用户数据查询只消费脱离 Session 的稳定快照。
"""
record = SiteUserDataSnapshot(
id=1,
domain="site.example",
name="示例站点",
username="tester",
userid="42",
user_level="Power User",
upload=1024.0,
download=512.0,
seeding=3.0,
leeching=1.0,
seeding_size=2048.0,
leeching_size=256.0,
)
repository = Mock()
repository.get_userdata_latest.return_value = [record]
result = SiteQueryService(repository).userdata_latest_sync()
assert isinstance(result[0], SiteUserData)
assert result[0].domain == "site.example"
assert result[0].name == "示例站点"
assert result[0].userid == "42"
assert result[0].upload == 1024