MoviePilot/tests/test_search_session.py

"""
订阅搜索检查点的任务租约、版本 CAS 与凭据排除回归。
"""

from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import create_engine, update
from sqlalchemy.orm import sessionmaker

from app.application.search.session import encode_search_state, torrent_snapshot
from app.db.adapters.searchsession import TransactionalSearchSessionRepository
from app.db.base import Base
from app.db.models.subscriptionsearch import SubscriptionSearchTask
from app.domain.context import TorrentInfo


@pytest.fixture
def repository(tmp_path):
    """
    创建隔离 SQLite 搜索库与有效任务租约，并在测试后释放引擎。
    """
    engine = create_engine(f"sqlite:///{tmp_path / 'search.db'}")
    Base.metadata.create_all(engine)
    now = datetime.now(timezone.utc).isoformat()
    expiry = (datetime.now(timezone.utc) + timedelta(minutes=15)).isoformat()
    with sessionmaker(bind=engine)() as session:
        session.add(SubscriptionSearchTask(task_id="task", batch_id="batch", subscription_id=1,
                    source="manual", position=0, state="running", created_at=now, updated_at=now,
                    lease_token="old", lease_expires_at=expiry))
        session.commit()
    yield TransactionalSearchSessionRepository(sessionmaker(bind=engine)), engine
    engine.dispose()


def _set_lease(engine, token):
    with sessionmaker(bind=engine)() as session:
        session.execute(update(SubscriptionSearchTask).values(lease_token=token))
        session.commit()


def test_only_current_task_lease_and_version_can_advance(repository):
    """
    只有当前任务租约与检查点版本匹配时才允许推进。
    """
    repo, engine = repository
    assert repo.create(task_id="task", payload="{}", task_lease="wrong") is None
    first = repo.create(task_id="task", payload='{"page":0}', task_lease="old")
    saved = repo.save(snapshot=first, payload='{"page":1}', task_lease="old")
    assert saved.version == 1
    # 迟到 worker 持有旧版本，不能覆盖新进度。
    assert repo.save(snapshot=first, payload='{"page":999}', task_lease="old") is None
    # 队列重新派发后旧租约立即失效，新租约无需等待即可继续。
    _set_lease(engine, "new")
    assert repo.save(snapshot=saved, payload='{"page":2}', task_lease="old") is None
    assert repo.save(snapshot=saved, payload='{"page":2}', task_lease="new").version == 2
    assert repo.get(task_id="task").payload == '{"page":2}'


def test_cancel_request_stops_checkpoint_writes(repository):
    """
    任务收到取消请求后停止写入检查点。
    """
    repo, engine = repository
    first = repo.create(task_id="task", payload="{}", task_lease="old")
    with sessionmaker(bind=engine)() as session:
        session.execute(update(SubscriptionSearchTask).values(cancel_requested=1))
        session.commit()
    assert repo.save(snapshot=first, payload="{}", task_lease="old") is None


def test_checkpoint_excludes_credentials_and_indirect_download_recipe():
    """
    检查点不保存凭据或可间接还原下载凭据的请求配方。
    """
    torrent = TorrentInfo(site=1, title="Show", site_cookie="cookie-secret",
                          enclosure="[encoded-api-key]https://api.example/download",
                          page_url="https://site.example/details.php?id=23")
    text = encode_search_state(torrent_snapshot(torrent))
    assert "cookie-secret" not in text
    assert "encoded-api-key" not in text
    assert '"enclosure": null' in text
    assert 'details.php?id=23' in text


def test_unrecognized_download_query_is_not_persisted():
    """
    无法确认安全性的下载查询参数不得持久化。
    """
    torrent = TorrentInfo(enclosure="https://site.example/download?id=1&passkey=secret")
    assert torrent_snapshot(torrent)["enclosure"] is None
    torrent.enclosure = "https://site.example/ticket/opaque-private-capability"
    assert torrent_snapshot(torrent)["enclosure"] is None


@pytest.mark.parametrize("system_stop,cancelled,expected", [(True, False, "requeued"), (False, True, "cancelled")])
def test_partial_submission_does_not_finish_an_interrupted_search_round(system_stop, cancelled, expected):
    """
    只提交部分目标不足以把中断的搜索轮次标记为完成。
    """
    from types import SimpleNamespace
    from unittest.mock import Mock

    from app.application.subscription.observability import finish_returned_search_task
    queue = Mock()
    execution = SimpleNamespace(download_started=True, incremental_search=True, scan_finished=False,
                                is_expired=lambda: False)
    _subscription, state, _reason = finish_returned_search_task(queue=queue, task_id="task", lease_token="lease",
         subscription_id=1, execution_context=execution, system_stopped=system_stop, cancel_requested=cancelled)
    assert state == expected
    queue.finish_task.assert_not_called()
    queue.release_task.assert_called_once()


def test_finished_round_removes_its_checkpoint_only_with_the_current_lease_and_version(repository):
    """
    完成轮次时仅允许当前租约与版本删除检查点。
    """
    repo, _ = repository
    first = repo.create(task_id="task", payload="{}", task_lease="old")
    repo.delete(snapshot=first, task_lease="stale")
    assert repo.get(task_id="task") is not None
    repo.delete(snapshot=first, task_lease="old")
    assert repo.get(task_id="task") is None


def test_stale_worker_cannot_delete_progress_saved_after_lease_handover(repository):
    # 旧执行者持有 version=0 快照；租约交接后新执行者保存 version=1，旧执行者随后删除不得生效。
    """
    租约交接后的旧工作任务不能删除新持有者保存的进度。
    """
    repo, engine = repository
    stale = repo.create(task_id="task", payload='{"page":0}', task_lease="old")
    _set_lease(engine, "new")
    fresh = repo.save(snapshot=stale, payload='{"page":1}', task_lease="new")
    assert fresh.version == 1
    repo.delete(snapshot=stale, task_lease="old")
    assert repo.get(task_id="task").version == 1
    # 即使旧执行者碰巧持有新租约，旧版本也删不掉新版本。
    repo.delete(snapshot=stale, task_lease="new")
    assert repo.get(task_id="task").version == 1


def _queue(tmp_path):
    from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository

    engine = create_engine(f"sqlite:///{tmp_path / 'lifecycle.db'}")
    Base.metadata.create_all(engine)
    return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine


@pytest.mark.parametrize("state", ["completed", "failed", "cancelled", "skipped"])
def test_terminal_task_releases_its_checkpoint(tmp_path, state):
    # 同一订阅之后的搜索是新任务，终态任务的检查点不会再被续用。
    """
    任务进入终态后释放关联检查点。
    """
    queue, engine = _queue(tmp_path)
    queue.enqueue(subscription_ids=(1,), source="timer", priority=10)
    task = queue.claim_next(owner="worker")
    assert queue.sessions.create(task_id=task.task_id, payload="{}", task_lease=task.lease_token)
    assert queue.finish_task(task_id=task.task_id, lease_token=task.lease_token, state=state, error=None)
    assert queue.sessions.get(task_id=task.task_id) is None
    engine.dispose()


def test_deferred_task_keeps_checkpoint_until_batch_is_cancelled(tmp_path):
    """
    延后执行的任务保留检查点，直到所属批次取消。
    """
    queue, engine = _queue(tmp_path)
    enqueued = queue.enqueue(subscription_ids=(1,), source="timer", priority=10)
    task = queue.claim_next(owner="worker")
    queue.sessions.create(task_id=task.task_id, payload="{}", task_lease=task.lease_token)
    assert queue.defer_task(task_id=task.task_id, lease_token=task.lease_token,
                            available_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
                            phase="waiting_site", message=None)
    assert queue.sessions.get(task_id=task.task_id) is not None
    # 排队中的任务被整批取消时直接进入终态，检查点同步删除。
    assert queue.request_cancel(enqueued.batch.batch_id)
    assert queue.sessions.get(task_id=task.task_id) is None
    engine.dispose()


def test_checkpoints_not_updated_for_fourteen_days_are_purged(repository):
    """
    十四天未更新的孤立检查点会被清理。
    """
    from app.db.models.searchsession import SearchSession

    repo, engine = repository
    now = datetime.now(timezone.utc)
    with sessionmaker(bind=engine)() as session:
        for task_id, days in (("orphan", 15), ("recent", 13)):
            session.add(SearchSession(task_id=task_id, version=0, payload="{}",
                                      updated_at=(now - timedelta(days=days)).isoformat(timespec="seconds")))
        session.commit()
    assert repo.create(task_id="task", payload="{}", task_lease="old")
    assert repo.get(task_id="orphan") is None
    assert repo.get(task_id="recent") is not None


@pytest.mark.parametrize("enclosure", [
    "https://site.example/download/0123456789abcdef0123456789abcdef?id=1",
    "https://site.example/dl/AbCdEfGh12345678IjKlMnOp9.torrent?id=1",
])
def test_passkey_in_download_path_is_not_persisted(enclosure):
    """
    下载路径中包含 passkey 时不保存该地址。
    """
    assert torrent_snapshot(TorrentInfo(enclosure=enclosure))["enclosure"] is None


def test_readable_page_paths_are_still_persisted():
    """
    可安全读取的详情页和下载路径仍允许持久化。
    """
    torrent = TorrentInfo(page_url="https://site.example/torrents/1234567890123456789",
                          enclosure="https://site.example/download.php?id=1&https=1")
    snapshot = torrent_snapshot(torrent)
    assert snapshot["page_url"] == torrent.page_url
    assert snapshot["enclosure"] == torrent.enclosure


def test_internal_candidate_id_is_not_exported():
    """
    内部候选标识不进入公开种子字典，并保持旧构造参数兼容。
    """
    torrent = TorrentInfo(site=1, title="Show")
    torrent.search_resource_id = "internal"
    assert "search_resource_id" not in torrent.to_dict()
    assert TorrentInfo(1).site == 1


def test_late_first_create_after_lease_handover_cannot_fail_the_new_owner(repository):
    # 审查复现的交错：旧执行者在租约交接后才插入首个检查点，新执行者此前读到的是空。
    """
    租约交接后迟到的首次创建不能覆盖或影响新持有者。
    """
    from app.db.models.searchsession import SearchSession

    repo, engine = repository
    _set_lease(engine, "new")
    # 租约条件在插入语句内校验：旧执行者交接后的首次创建不落库。
    assert repo.create(task_id="task", payload='{"by":"old"}', task_lease="old") is None
    assert repo.get(task_id="task") is None
    # 即便旧行已经写入（如交接前已通过校验的迟到事务），新执行者也以版本递增接管，不因唯一键失败。
    with sessionmaker(bind=engine)() as session:
        session.add(SearchSession(task_id="task", version=0, payload='{"by":"old"}',
                                  updated_at=datetime.now(timezone.utc).isoformat(timespec="seconds")))
        session.commit()
    stale = repo.get(task_id="task")
    owned = repo.create(task_id="task", payload='{"by":"new"}', task_lease="new")
    assert (owned.version, owned.payload) == (1, '{"by":"new"}')
    # 旧执行者既不能再创建覆盖，也不能用旧版本推进或删除。
    assert repo.create(task_id="task", payload='{"by":"old"}', task_lease="old") is None
    assert repo.save(snapshot=stale, payload='{"by":"old"}', task_lease="old") is None
    repo.delete(snapshot=stale, task_lease="old")
    assert repo.get(task_id="task").payload == '{"by":"new"}'


def test_first_create_statement_compiles_for_postgresql():
    """
    首次创建检查点使用 PostgreSQL 可编译的冲突忽略语句。
    """
    from sqlalchemy import literal, select
    from sqlalchemy.dialects import postgresql
    from sqlalchemy.dialects.postgresql import insert

    from app.db.models.searchsession import SearchSession

    statement = insert(SearchSession).from_select(
        ["task_id", "version", "payload", "updated_at"], select(literal("t"), literal(0), literal("{}"), literal("now")),
    ).on_conflict_do_nothing(index_elements=[SearchSession.task_id])
    assert "ON CONFLICT (task_id) DO NOTHING" in str(statement.compile(dialect=postgresql.dialect()))