mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
389 lines
18 KiB
Python
389 lines
18 KiB
Python
"""
|
|
订阅业务活动的事务、隔离、计数与等待语义回归测试。
|
|
"""
|
|
|
|
import asyncio
|
|
from contextlib import contextmanager
|
|
from dataclasses import replace
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.application.subscription.activity import (
|
|
SubscriptionActivityRun,
|
|
SubscriptionActivityService,
|
|
current_activity_run,
|
|
queued_activity,
|
|
)
|
|
from app.application.subscription.contract import (
|
|
SubscriptionHistoryPatch,
|
|
SubscriptionIdentity,
|
|
SubscriptionPatch,
|
|
SubscriptionSnapshot,
|
|
)
|
|
from app.application.subscription.sitebudget import SubscriptionSearchDeferred
|
|
from app.db.adapters.subscription import SessionSubscriptionRepository
|
|
from app.db.adapters.subscriptionactivity import TransactionalSubscriptionActivityRepository
|
|
from app.db.models.subscribe import Subscribe
|
|
from app.db.models.subscribeactivity import SubscribeActivity
|
|
from app.db.models.subscribehistory import SubscribeHistory
|
|
from app.db.session import async_session_scope
|
|
from app.schemas.types import MediaSource
|
|
|
|
|
|
def _snapshot(**values) -> SubscriptionSnapshot:
|
|
"""
|
|
创建拥有独立活动身份的测试订阅。
|
|
"""
|
|
return SubscriptionSnapshot(id=1, name="Example", activity_key=uuid4().hex, resolution="2160p", **values)
|
|
|
|
|
|
def _resource(title="Example.S01E09", site=1):
|
|
"""
|
|
生成含敏感下载地址的候选，验证活动摘要不会泄漏该地址。
|
|
"""
|
|
return SimpleNamespace(torrent_info=SimpleNamespace(
|
|
title=title, site=site, site_name=f"Site {site}",
|
|
enclosure="https://example.test/download?passkey=private", site_cookie="private",
|
|
))
|
|
|
|
|
|
def test_creation_and_activity_rollback_together(db):
|
|
"""
|
|
未提交的订阅与创建活动必须一起回滚，重复创建不会重复记录。
|
|
"""
|
|
db.watermark(Subscribe, SubscribeActivity)
|
|
repository = SessionSubscriptionRepository(db.session)
|
|
identity = SubscriptionIdentity(media_source=MediaSource.TMDB, media_id="9916963", type="电视剧", season=1)
|
|
payload = SubscriptionPatch({"name": "Example", "media_source": "tmdb", "media_id": "9916963", "type": "电视剧", "season": 1, "username": "owner"})
|
|
result = repository.stage_add(identity, payload, "owner")
|
|
assert result.created
|
|
event_key = db.session.execute(select(SubscribeActivity.activity_key).where(SubscribeActivity.kind == "created")).scalars().all()[-1]
|
|
key = repository.get(result.subscribe_id).activity_key
|
|
assert event_key == key
|
|
db.session.rollback()
|
|
assert db.session.get(Subscribe, result.subscribe_id) is None
|
|
assert not db.session.execute(select(SubscribeActivity).where(SubscribeActivity.activity_key == key)).scalars().all()
|
|
created = repository.stage_add(identity, payload, "owner")
|
|
db.session.commit()
|
|
assert not repository.stage_add(identity, payload, "owner").created
|
|
assert len(db.session.execute(select(SubscribeActivity).where(SubscribeActivity.activity_key == repository.get(created.subscribe_id).activity_key)).scalars().all()) == 1
|
|
|
|
|
|
def test_updates_freeze_rules_and_history_keeps_original_identity(db):
|
|
"""
|
|
人工修改保存前后值，自动时间更新不刷屏，完成历史关联原订阅。
|
|
"""
|
|
db.watermark(Subscribe, SubscribeHistory, SubscribeActivity)
|
|
subscribe = Subscribe(name="Example", username="owner", media_source="tmdb", media_id="9916964", resolution="1080p")
|
|
db.add(subscribe)
|
|
repository = SessionSubscriptionRepository(db.session)
|
|
repository.stage_update(subscribe.id, SubscriptionPatch({"last_update": "2026-10-12"}))
|
|
assert not db.session.execute(select(SubscribeActivity).where(SubscribeActivity.activity_key == subscribe.activity_key)).scalars().all()
|
|
snapshot = repository.stage_update(subscribe.id, SubscriptionPatch({"resolution": "2160p"}))
|
|
repository.stage_history(SubscriptionHistoryPatch.from_subscription(snapshot.to_dict()))
|
|
repository.stage_delete_sync(subscribe.id)
|
|
db.session.commit()
|
|
history = db.session.execute(select(SubscribeHistory).where(SubscribeHistory.activity_key == snapshot.activity_key)).scalar_one()
|
|
events = db.session.execute(select(SubscribeActivity).where(SubscribeActivity.activity_key == history.activity_key).order_by(SubscribeActivity.id)).scalars().all()
|
|
assert [event.kind for event in events] == ["updated", "completed"]
|
|
assert events[0].details["changes"] == [{"field": "resolution", "before": "1080p", "after": "2160p"}]
|
|
activity_repository = TransactionalSubscriptionActivityRepository(lambda: db.session, async_session_scope)
|
|
history_id, event_id = history.id, events[0].event_id
|
|
|
|
async def verify_history_access():
|
|
"""
|
|
完成后仍可按历史 owner 读取原始活动和规则快照。
|
|
"""
|
|
page = await activity_repository.list_accessible(history_id, "owner", False, history=True,
|
|
kind=None, since=None, before=None, limit=30)
|
|
assert page and len(page[0]) == 2
|
|
detail = await activity_repository.get_accessible(history_id, event_id, "owner", False, history=True)
|
|
assert detail and detail.details["rules"]["resolution"] == "2160p"
|
|
assert await activity_repository.get_accessible(history_id, event_id, "other", False, history=True) is None
|
|
|
|
asyncio.run(verify_history_access())
|
|
replacement = Subscribe(id=subscribe.id, name="Replacement", username="owner", media_source="tmdb", media_id="9916965")
|
|
db.add(replacement)
|
|
assert replacement.activity_key != history.activity_key
|
|
|
|
|
|
def test_round_counts_are_distinct_and_details_do_not_include_credentials():
|
|
"""
|
|
候选、符合条件和实际提交分别去重，重复筛选不会夸大计数。
|
|
"""
|
|
run = SubscriptionActivityRun(_snapshot(), "search", "manual")
|
|
resource = _resource()
|
|
run.resource(resource, "related")
|
|
run.resource(resource, "rejected", "不符合分辨率")
|
|
run.resource(resource, "rejected", "不符合优先级")
|
|
assert run.details["resources"][0]["reason"] == "不符合优先级"
|
|
run.resource(resource, "qualified")
|
|
run.resource(resource, "qualified")
|
|
run.resource(resource, "submitted")
|
|
run.resource(resource, "qualified")
|
|
run.resource(resource, "submitted")
|
|
assert (run.counts["related"], run.counts["qualified"], run.counts["submitted"]) == (1, 1, 1)
|
|
assert "private" not in str(run.snapshot())
|
|
run.site(1, candidates=32)
|
|
run.site(2, status="waiting")
|
|
assert run.counts["requested_sites"] == 1
|
|
assert run.counts["discovered"] == 32
|
|
|
|
|
|
def test_waiting_is_not_failure_and_scope_is_released():
|
|
"""
|
|
暂时站点冲突保存继续时间，不伪装成失败或完成。
|
|
"""
|
|
saved = []
|
|
service = SubscriptionActivityService(SimpleNamespace(save=saved.append))
|
|
with pytest.raises(SubscriptionSearchDeferred):
|
|
with service.scope(_snapshot(), "search", "automatic"):
|
|
raise SubscriptionSearchDeferred(retry_at="2026-10-12T10:50:00+00:00", site_ids=(8,), wait_reason="busy")
|
|
assert saved[-1].status == "waiting"
|
|
assert saved[-1].details["pending_site_ids"] == [8]
|
|
assert current_activity_run() is None
|
|
with service.scope(_snapshot(), "refresh", "automatic"):
|
|
pass
|
|
assert saved[-1].status == "empty"
|
|
|
|
|
|
def test_activity_details_are_limited_without_losing_summary_counts():
|
|
"""
|
|
大量资源只保留有限摘要，业务总数仍覆盖整轮。
|
|
"""
|
|
run = SubscriptionActivityRun(_snapshot(), "refresh", "automatic")
|
|
for index in range(130):
|
|
run.resource(_resource(f"Example.{index}"), "rejected", "不符合条件")
|
|
assert run.counts["related"] == 130
|
|
assert len(run.details["resources"]) == 100
|
|
assert run.details["resources_omitted"] == 30
|
|
|
|
|
|
def test_search_retries_continue_the_same_event_without_duplicate_counts():
|
|
"""
|
|
自动恢复关联原搜索，并保留每次处理结果而不重复统计下载。
|
|
"""
|
|
stored = {}
|
|
|
|
def save(event):
|
|
"""
|
|
模拟固定事件身份的持久化更新。
|
|
"""
|
|
stored[event.event_id] = event
|
|
|
|
def get(activity_key, event_id):
|
|
"""
|
|
只返回当前订阅拥有的历史轮次。
|
|
"""
|
|
event = stored.get(event_id)
|
|
return event if event is not None and event.activity_key == activity_key else None
|
|
|
|
subscribe = _snapshot()
|
|
service = SubscriptionActivityService(SimpleNamespace(save=save, get=get))
|
|
with pytest.raises(SubscriptionSearchDeferred):
|
|
with service.scope(subscribe, "search", "automatic", "task-6963") as run:
|
|
run.resource(_resource(), "submitted")
|
|
raise SubscriptionSearchDeferred(retry_at="2026-10-12T10:50:00+00:00", site_ids=(8,))
|
|
first_started = stored["task-6963"].started_at
|
|
with service.scope(subscribe, "search", "automatic", "task-6963") as run:
|
|
run.resource(_resource(), "qualified")
|
|
run.resource(_resource(), "submitted")
|
|
assert len(stored) == 1
|
|
event = stored["task-6963"]
|
|
assert event.started_at == first_started
|
|
assert event.status == "completed"
|
|
assert (event.counts["related"], event.counts["qualified"], event.counts["submitted"]) == (1, 1, 1)
|
|
assert [attempt["status"] for attempt in event.details["attempts"]] == ["waiting", "completed"]
|
|
|
|
|
|
def test_activity_query_enforces_owner_history_and_stable_pagination(db):
|
|
"""
|
|
普通用户不能跨订阅或游标读取，新增事件不会导致翻页重复。
|
|
"""
|
|
db.watermark(Subscribe, SubscribeHistory, SubscribeActivity)
|
|
subscribe = Subscribe(name="Example", username="owner", media_source="tmdb", media_id="9916966")
|
|
db.add(subscribe)
|
|
|
|
@contextmanager
|
|
def sync_scope():
|
|
"""
|
|
测试写入复用已隔离的数据库会话。
|
|
"""
|
|
yield db.session
|
|
|
|
repository = TransactionalSubscriptionActivityRepository(sync_scope, async_session_scope)
|
|
snapshot = _snapshot(username="owner")
|
|
snapshot = replace(snapshot, id=subscribe.id, activity_key=subscribe.activity_key)
|
|
for _ in range(3):
|
|
repository.save(SubscriptionActivityRun(snapshot, "search", "manual").snapshot())
|
|
|
|
async def query():
|
|
"""
|
|
执行授权与游标边界断言。
|
|
"""
|
|
params = dict(history=False, kind=None, since=None, before=None, limit=2)
|
|
assert await repository.list_accessible(subscribe.id, "other", False, **params) is None
|
|
first, cursor = await repository.list_accessible(subscribe.id, "owner", False, **params)
|
|
assert len(first) == 2 and cursor
|
|
repository.save(SubscriptionActivityRun(snapshot, "search", "manual").snapshot())
|
|
last, next_cursor = await repository.list_accessible(subscribe.id, "owner", False, **{**params, "before": cursor})
|
|
assert len(last) == 1 and next_cursor is None
|
|
assert {event.event_id for event in first}.isdisjoint(event.event_id for event in last)
|
|
assert await repository.list_accessible(subscribe.id, "other", True, **params)
|
|
|
|
asyncio.run(query())
|
|
|
|
|
|
def test_failed_submission_is_distinct_from_empty_and_recovery():
|
|
"""
|
|
下载器拒绝是真实失败，恢复成功后清除失败计数而不重复关联资源。
|
|
"""
|
|
saved = []
|
|
service = SubscriptionActivityService(SimpleNamespace(save=saved.append))
|
|
with service.scope(_snapshot(), "refresh", "automatic") as run:
|
|
run.resource(_resource(), "qualified")
|
|
run.resource(_resource(), "failed", "下载器未接受该资源，未提交成功")
|
|
assert saved[-1].status == "failed"
|
|
assert saved[-1].counts["download_failed"] == 1
|
|
with service.scope(_snapshot(), "refresh", "automatic") as run:
|
|
run.resource(_resource(), "failed")
|
|
run.resource(_resource(), "submitted")
|
|
run.resource(_resource(), "failed")
|
|
assert saved[-1].status == "completed"
|
|
assert saved[-1].counts["related"] == saved[-1].counts["submitted"] == 1
|
|
assert saved[-1].counts["download_failed"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_details_cannot_cross_subscription_or_owner(db):
|
|
"""
|
|
猜中事件 ID 也不能跨 owner、活动整数 ID 或历史整数 ID 读取详情。
|
|
"""
|
|
db.watermark(Subscribe, SubscribeActivity)
|
|
owned = Subscribe(name="Owned", username="owner")
|
|
other = Subscribe(name="Other", username="other")
|
|
db.add(owned, other)
|
|
repository = TransactionalSubscriptionActivityRepository(lambda: db.session, async_session_scope)
|
|
snapshot = replace(_snapshot(), id=owned.id, activity_key=owned.activity_key)
|
|
owned_id, other_id = owned.id, other.id
|
|
event = SubscriptionActivityRun(snapshot, "refresh", "automatic").snapshot()
|
|
repository.save(event)
|
|
assert await repository.get_accessible(owned_id, event.event_id, "owner", False, history=False)
|
|
assert await repository.get_accessible(owned_id, event.event_id, "admin", True, history=False)
|
|
assert await repository.get_accessible(owned_id, event.event_id, "other", False, history=False) is None
|
|
assert await repository.get_accessible(other_id, event.event_id, "admin", True, history=False) is None
|
|
assert await repository.get_accessible(owned_id, event.event_id, "owner", False, history=True) is None
|
|
|
|
|
|
def test_all_failed_sites_are_not_reported_as_an_empty_success():
|
|
"""
|
|
全部站点失败与正常零结果分开，未发出的请求不进入实际请求计数。
|
|
"""
|
|
saved = []
|
|
service = SubscriptionActivityService(SimpleNamespace(save=saved.append))
|
|
with service.scope(_snapshot(), "search", "automatic") as run:
|
|
run.counts["sites"] = 2
|
|
run.site(1, status="failed", attempted=True)
|
|
run.site(2, status="failed", attempted=True)
|
|
assert saved[-1].status == "failed" and saved[-1].counts["failed_sites"] == 2
|
|
with service.scope(_snapshot(), "search", "automatic") as run:
|
|
run.site(1, status="received", candidates=1, attempted=False)
|
|
run.site(2, status="failed", attempted=True)
|
|
assert saved[-1].status == "empty" and saved[-1].counts["requested_sites"] == 1
|
|
assert "1 个站点" in saved[-1].details["reason"]
|
|
|
|
|
|
def test_queued_rules_are_replaced_by_actual_execution_rules():
|
|
"""
|
|
入队和执行之间的编辑必须使用真正执行时的条件，原请求时间仍保留。
|
|
"""
|
|
subscribe = _snapshot()
|
|
queued = queued_activity(replace(subscribe, resolution="1080p"), "task-queued", "new", None)
|
|
saved = []
|
|
service = SubscriptionActivityService(SimpleNamespace(save=saved.append, get=lambda *_args: queued))
|
|
with service.scope(subscribe, "search", "manual", "task-queued"):
|
|
pass
|
|
assert saved[-1].started_at == queued.started_at
|
|
assert saved[-1].details["rules"]["resolution"] == "2160p"
|
|
assert saved[-1].details["attempts"][0]["rules"]["resolution"] == "2160p"
|
|
|
|
|
|
def test_refresh_failure_is_visible_without_claiming_a_match(monkeypatch):
|
|
"""
|
|
候选刷新失败会记录业务失败，暂停和其他媒体类型不会被误记为已检查。
|
|
"""
|
|
from app.chain.subscribe.refresh import SubscribeRefreshOwner
|
|
|
|
saved = []
|
|
service = SubscriptionActivityService(SimpleNamespace(save=saved.append))
|
|
owner = object.__new__(SubscribeRefreshOwner)
|
|
owner.subscription_activity = service
|
|
owner.subscription_repository = SimpleNamespace(list=lambda: [
|
|
replace(_snapshot(), state="R", type="电视剧"),
|
|
replace(_snapshot(), id=2, state="S", type="电视剧"),
|
|
replace(_snapshot(), id=3, state="R", type="电影"),
|
|
])
|
|
owner.get_subscribed_sites = lambda _type: [1]
|
|
owner.has_music_subscribe = lambda _type: False
|
|
|
|
def fail_refresh(*_args, **_kwargs):
|
|
"""
|
|
模拟刷新未完成，未调用外部站点。
|
|
"""
|
|
raise RuntimeError("offline source failed")
|
|
|
|
monkeypatch.setattr("app.chain.subscribe.refresh.TorrentsChain", lambda: SimpleNamespace(refresh=fail_refresh))
|
|
with pytest.raises(RuntimeError, match="offline source"):
|
|
owner.refresh(mtype="电视剧", source="manual")
|
|
assert len(saved) == 2 and saved[-1].source == "manual"
|
|
assert saved[-1].status == "failed" and saved[-1].details["phase"] == "refresh_failed"
|
|
assert saved[-1].counts["related"] == saved[-1].counts["submitted"] == 0
|
|
|
|
|
|
def test_bounded_details_keep_actual_download_receipts():
|
|
"""
|
|
大量拒绝候选不能挤掉后出现的下载回执，同时完整相关资源计数不丢失。
|
|
"""
|
|
run = SubscriptionActivityRun(_snapshot(), "search", "manual")
|
|
for index in range(130):
|
|
run.resource(_resource(f"Rejected.{index}"), "rejected", "不符合条件")
|
|
run.resource(_resource("Accepted.S01E09"), "submitted")
|
|
snapshot = run.snapshot()
|
|
assert snapshot.counts["related"] == 131 and snapshot.counts["submitted"] == 1
|
|
assert len(snapshot.details["resources"]) == 100
|
|
assert snapshot.details["resources"][0]["title"] == "Accepted.S01E09"
|
|
assert snapshot.details["resources_omitted"] == 31
|
|
|
|
|
|
def test_resumed_site_request_keeps_the_original_planned_denominator():
|
|
"""
|
|
七站点已请求、一个等待的搜索恢复后变为八比八，不能变为八比一。
|
|
"""
|
|
stored = {}
|
|
|
|
def save(event):
|
|
"""
|
|
按固定任务身份持久化当前快照。
|
|
"""
|
|
stored[event.event_id] = event
|
|
|
|
subscribe = _snapshot()
|
|
service = SubscriptionActivityService(SimpleNamespace(save=save, get=lambda *_args: stored.get("resumed-sites")))
|
|
with pytest.raises(SubscriptionSearchDeferred):
|
|
with service.scope(subscribe, "search", "automatic", "resumed-sites") as run:
|
|
run.plan_sites(range(1, 9))
|
|
for site_id in range(1, 8):
|
|
run.site(site_id, candidates=1)
|
|
run.site(8, status="waiting")
|
|
raise SubscriptionSearchDeferred(retry_at="2026-10-12T10:50:00Z", site_ids=(8,))
|
|
assert stored["resumed-sites"].counts["requested_sites"] == 7
|
|
with service.scope(subscribe, "search", "automatic", "resumed-sites") as run:
|
|
run.plan_sites((8,))
|
|
run.site(8, candidates=1)
|
|
event = stored["resumed-sites"]
|
|
assert event.counts["sites"] == event.counts["requested_sites"] == 8
|
|
assert event.counts["discovered"] == 8 and event.counts["waiting_sites"] == 0
|