From f0b34cf9f5daab7000660049f5dffd3376881802 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 8 Oct 2026 11:57:10 +0800 Subject: [PATCH] fix(search): restore concurrent progressive manual paging --- app/chain/search/manual.py | 90 +++++++++++++++------ docs/mcp-api.md | 2 +- docs/smart-search.md | 1 + skills/moviepilot-api/SKILL.md | 4 +- tests/test_search_manual.py | 139 +++++++++++++++++++++++++++++++-- 5 files changed, 204 insertions(+), 32 deletions(-) diff --git a/app/chain/search/manual.py b/app/chain/search/manual.py index 4effce705..f52c122cb 100644 --- a/app/chain/search/manual.py +++ b/app/chain/search/manual.py @@ -4,14 +4,15 @@ import asyncio import base64 import binascii import json +from contextlib import aclosing from functools import partial -from typing import Any, AsyncIterator, Callable, Optional, TypeVar, cast +from typing import Any, AsyncGenerator, AsyncIterator, Callable, Optional, TypeVar, cast from uuid import uuid4 from app.application.configuration import get_configured_system_config from app.chain.media import MediaChain from app.chain.search.contract import _SearchOwnerBase -from app.chain.search.execution import MediaSearchPlan +from app.chain.search.execution import MediaSearchPlan, _candidate_contexts from app.chain.search.media import _build_missing_media_map from app.chain.search.result import SearchResultOwner from app.chain.search.scan import RESTART_AFTER_SECONDS, SearchSources @@ -59,6 +60,7 @@ class _ManualPage: """一次单页请求只收集原始资源，结果处理委托给原搜索业务。""" def __init__(self, owner: _SearchOwnerBase, plan: MediaSearchPlan, raw_title: bool, search_id: str) -> None: + """持有本轮搜索计划和页摘要，候选只由事件编排层顺序收集。""" self.owner, self.plan, self.raw_title = owner, plan, raw_title self.media = cast(MediaInfo, plan.mediainfo) self.sources = SearchSources(owner, plan) @@ -66,7 +68,7 @@ class _ManualPage: self.search_id = search_id self.page_cache = Cache(ttl=RESTART_AFTER_SECONDS) - def read(self, key: str, page: int) -> dict[str, Any]: + def read(self, key: str, page: int) -> tuple[dict[str, Any], list[TorrentInfo]]: """上一页摘要保留 36 小时；失效时先补上一页再取目标页，只有目标页交给原结果处理。""" query = self.sources.queries[key] pageable = self.sources.pageable(key) @@ -76,7 +78,7 @@ class _ManualPage: "page": page, "can_continue": False, "error": None} # 超过最大页数，或不支持分页的站点请求后续页时，直接按末页返回。 if page >= MAX_SEARCH_PAGES or (page > 0 and not pageable): - return facts + return facts, [] previous = self.page_cache.get(f"{source}:{page - 1}", region=_PAGE_CACHE_REGION) if page else None pages = [page - 1, page] if page and previous is None else [page] for requested_page in pages: @@ -86,33 +88,61 @@ class _ManualPage: if not (observation.attempted and observation.outcome == "success"): facts.update(error=observation.error or "本页请求未完成，请继续加载重试", can_continue=query["site"] != "plugin") - return facts + return facts, [] signature = search_page_signature(search_resource_id(item) for item in result.torrents) or "" self.page_cache.set(f"{source}:{requested_page}", signature, region=_PAGE_CACHE_REGION) if requested_page < page: previous = signature raw_count = observation.raw_count if observation.raw_count is not None else len(result.torrents) if previous and signature == previous: - return facts + return facts, [] # 尚未确认末页时允许尝试下一页，不预取，也不保证下一页一定有结果。 facts.update(can_continue=bool( pageable and raw_count and observation.has_more is not False and page + 1 < MAX_SEARCH_PAGES)) - self.torrents.extend(result.torrents) - return facts + return facts, result.torrents - def results(self) -> list[Context]: + def results(self, torrents: Optional[list[TorrentInfo]] = None) -> list[Context]: """复用原搜索的识别、季集过滤和排序；显示过滤与分页能力无关。""" + torrents = self.torrents if torrents is None else torrents if self.raw_title: return SearchTitleOwner._resolve_title_request(cast(SearchTitleOwner, self.owner), _TitleSearchResolveRequest( - title=self.media.title, torrents=self.torrents, rule_groups=self.plan.rule_groups, + title=self.media.title, torrents=torrents, rule_groups=self.plan.rule_groups, mtype=None if self.media.type == MediaType.UNKNOWN else self.media.type, )).contexts return SearchResultOwner._parse_result( - cast(SearchResultOwner, self.owner), torrents=self.torrents, mediainfo=self.media, keyword=self.plan.keyword, + cast(SearchResultOwner, self.owner), torrents=torrents, mediainfo=self.media, keyword=self.plan.keyword, season_episodes=self.sources.season_episodes, rule_groups=self.plan.rule_groups, custom_words=self.plan.custom_words, filter_params=self.plan.filter_params, ) + def preview(self, torrents: list[TorrentInfo]) -> list[Context]: + """标题预览复用过滤，精确搜索先显示未匹配候选，最终整页统一校验。""" + return self.results(torrents) if self.raw_title else _candidate_contexts(self.media, torrents) + + +async def _read_pages( + manual: _ManualPage, keys: list[str], page: int, +) -> AsyncGenerator[tuple[str, dict[str, Any], list[TorrentInfo]], None]: + """复用有界并发和取消收口；同站点多个关键词按轮次串行，先完成的页立即交付。""" + grouped: dict[str, list[str]] = {} + for key in keys: + grouped.setdefault(manual.sources.queries[key]["site"], []).append(key) + + async def read_source(site: dict[str, Any], requested_page: int) -> list[Any]: + """把指定来源的页事实适配到既有单页调度器。""" + return [await _offload(manual.read, site["key"], requested_page)] + + for index in range(max(map(len, grouped.values()), default=0)): + pages = manual.owner._iter_site_page_results( + indexer_sites=[{"key": group[index]} for group in grouped.values() if index < len(group)], + search_pages=[page], search_page=read_source, should_continue=lambda _site, _items: False, + task_owner="chain.search.manual.site_page", + ) + async with aclosing(pages): + async for site, _page, results, _continued, _elapsed in pages: + facts, torrents = results[0] + yield site["key"], facts, torrents + async def _prepare_page(owner: _SearchOwnerBase, params: dict[str, Any]) -> _ManualPage: """只构造本次请求的识别/关键词计划，不读写任何搜索检查点。""" @@ -139,7 +169,7 @@ async def _prepare_page(owner: _SearchOwnerBase, params: dict[str, Any]) -> _Man return await _offload(_ManualPage, owner, plan, raw_title, search_id) -async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIterator[dict[str, Any]]: +async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]: """每个来源只返回指定页；只有首页请求才使用别名回退和插件来源。""" page = max(0, int(params.get("page") or 0)) initial = not params.get("source") @@ -150,15 +180,27 @@ async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIter keys = list(manual.sources.queries) else: keys = manual.sources.add_keywords(all_names=bool(manual.owner.runtime_config.search_multiple_name)) - facts: list[dict[str, Any]] = [] + facts_by_key: dict[str, dict[str, Any]] = {} + torrents_by_key: dict[str, list[TorrentInfo]] = {} contexts: list[Context] = [] + preview_count = 0 while keys: - for key in keys: - if key.startswith("plugin:") and not initial: - continue - facts.append(await _offload(manual.read, key, page)) - yield {"type": "progress", "stage": "searching", "value": 90, - "text": "已请求一页，正在处理其他搜索来源"} + pages = _read_pages(manual, keys, page) + async with aclosing(pages): + async for key, facts, torrents in pages: + facts_by_key[key], torrents_by_key[key] = facts, torrents + manual.torrents.extend(torrents) + previews = await _offload(manual.preview, torrents) + preview_count += len(previews) + yield {"type": "append", "stage": "searching", + "value": len(facts_by_key) / len(manual.sources.queries) * 90, + "text": f"已完成 {len(facts_by_key)} / {len(manual.sources.queries)} 个搜索来源", + "items": [context.to_dict() for context in previews], + "total_items": preview_count, "candidate_items": len(manual.torrents)} + # 并发完成顺序只影响预览，完整结果仍保持原来源顺序和最终排序语义。 + manual.torrents = [torrent for key in manual.sources.queries for torrent in torrents_by_key.get(key, [])] + yield {"type": "progress", "stage": "filtering", "value": 98, + "text": f"正在过滤匹配 {len(manual.torrents)} 个候选资源 ..."} contexts = await _offload(manual.results) if contexts or not initial: break @@ -173,9 +215,10 @@ async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIter area=manual.plan.area, season=None if manual.raw_title else params.get("season"), sites=params.get("sites"), result_type="torrent") items = [context.to_dict() for context in contexts] + source_facts = [facts_by_key[key] for key in manual.sources.queries if key in facts_by_key] yield {"type": "replace", "stage": "filtered", "value": 100, "items": items, - "total_items": len(items), "candidate_items": len(manual.torrents), "sources": facts} - yield {"type": "done", "stage": "done", "total_items": len(items), "sources": facts} + "total_items": len(items), "candidate_items": len(manual.torrents), "sources": source_facts} + yield {"type": "done", "stage": "done", "total_items": len(items), "sources": source_facts} class SearchManualOwner: @@ -189,5 +232,6 @@ class SearchManualOwner: except ValueError as error: yield {"type": "error", "message": str(error)} return - async for event in _page_events(manual, params): - yield event + async with aclosing(_page_events(manual, params)) as events: + async for event in events: + yield event diff --git a/docs/mcp-api.md b/docs/mcp-api.md index ac787eb74..ed4bd3d32 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -1174,4 +1174,4 @@ description、aliases、instructions，或通过 `append_instructions` 追加规 ## Web 手动分页 -`/api/v1/search/title/stream` 与 `/api/v1/search/media/{media_id}/stream` 新增可选参数 `manual_paging`、`page`、`source`。`manual_paging=true` 时每个来源只返回指定的一页，`replace`/`done` 事件附带各来源的 `source`、`site_name`、`page`、`can_continue`、`error`。不传该参数时行为不变，Agent `search.*` operation 不受影响。完整规则及订阅补全搜索策略 `SubscribeSearchStrategy` 见 [订阅补全搜索与手动分页](smart-search.md)。 +`/api/v1/search/title/stream` 与 `/api/v1/search/media/{media_id}/stream` 新增可选参数 `manual_paging`、`page`、`source`。`manual_paging=true` 时每个来源只返回指定的一页，不同站点按既有并发数请求，并在来源完成后发送 `append` 预览；最终统一过滤后的 `replace`/`done` 事件附带各来源的 `source`、`site_name`、`page`、`can_continue`、`error`，预览不推进页号。不传该参数时行为不变，Agent `search.*` operation 不受影响。完整规则及订阅补全搜索策略 `SubscribeSearchStrategy` 见 [订阅补全搜索与手动分页](smart-search.md)。 diff --git a/docs/smart-search.md b/docs/smart-search.md index 48d4fa44d..4931ebf6b 100644 --- a/docs/smart-search.md +++ b/docs/smart-search.md @@ -36,6 +36,7 @@ `/search/title/stream` 和 `/search/media/{id}/stream` 传 `manual_paging=true` 时启用单页模式： - 首次请求（不带 `source`）每个来源请求第 0 页，只有这时才做别名回退和插件搜索。 +- 不同站点按既有搜索并发数请求，同站点的多个关键词依次请求；每个来源完成后立即发送 `append` 预览，慢站点不会阻塞已完成来源的展示。标题预览遵循搜索过滤规则，精确媒体搜索先展示未匹配候选，最终结果统一匹配、过滤和排序。 - `replace` 和 `done` 事件带 `sources` 列表，每项包含不透明的 `source`、`site_name`、本次 `page`、`can_continue` 和 `error`。 - 续页时只回传 `source` 和要请求的 `page`（零基）；客户端在整页结果应用后才推进页号，失败时保留页号以便重试。 - 后端按本轮来源和页号缓存原始页摘要 36 小时，用于判断与上一页是否重复；上一页摘要缺失时先补请求上一页。空页、重复页、明确末页、不支持分页的站点或达到 100 页时 `can_continue=false`。`can_continue=true` 只表示可以继续尝试，后端不预取、不自动重试。 diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 76d037b94..5322dd7e6 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -275,7 +275,9 @@ The `download.add` body must contain `torrent_in` (at least `title` and `enclosu ## Collection Counts And Pagination - Search SSE `manual_paging=true` is a Web-only page-by-page mode and adds no Agent - operations; keep using the documented `search.*` operations. + operations; keep using the documented `search.*` operations. Web streams search + sites concurrently and emit `append` previews before the final filtered page; + only the completed result carries source state for advancing page numbers. - For list inspection, explicitly send the operation's documented pagination fields instead of requesting an unbounded legacy result. For optional legacy pagination, start with `query={"page":1,"count":20}`. diff --git a/tests/test_search_manual.py b/tests/test_search_manual.py index 050b68f5d..1e816e3ab 100644 --- a/tests/test_search_manual.py +++ b/tests/test_search_manual.py @@ -1,6 +1,10 @@ """Web 手动分页：后端缓存页摘要并恢复过期比较依据，页号与重试由客户端持有。""" +import asyncio +import threading +from dataclasses import replace from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest @@ -8,17 +12,19 @@ from app.application.site.observation import report_site_search_outcome, report_ from app.chain.search import manual as manual_module from app.chain.search.facade import SearchChain from app.chain.search.manual import SearchManualOwner, decode_source, encode_source -from app.domain.context import TorrentInfo -from app.schemas.types import MediaType +from app.domain.context import MediaInfo, TorrentInfo +from app.schemas.types import MediaSource, MediaType @pytest.fixture def page_clock(monkeypatch): + """使用可控时钟隔离页摘要缓存，验证失效后的上一页补取。""" from app.runtime import cache as cache_module clock = [0.0] original = cache_module.MemoryTLRUCache.__init__ def initialize(cache, *args, **kwargs): + """让测试缓存的过期判断统一使用可控时钟。""" original(cache, *args, **kwargs, timer=lambda: clock[0]) monkeypatch.setattr(cache_module.MemoryTLRUCache, "__init__", initialize) monkeypatch.setattr(cache_module.MemoryBackend, "_region_caches", {}) @@ -28,6 +34,7 @@ def page_clock(monkeypatch): @pytest.fixture def manual_owner(monkeypatch, page_clock): + """构造离线搜索来源和保存边界，记录页请求并模拟失败。""" owner = SearchChain() sites = [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}] monkeypatch.setattr(owner, "_sync_indexers", lambda ids: [site for site in sites if not ids or site["id"] in ids]) @@ -36,11 +43,13 @@ def manual_owner(monkeypatch, page_clock): saved = [] monkeypatch.setattr(owner, "save_last_search_params", lambda **params: saved.append(params)) async def save_results(contexts): + """记录首次搜索的保存行为，避免写入真实缓存。""" saved.append(len(contexts)) monkeypatch.setattr(owner, "_async_save_results", save_results) calls = [] failures = set() def request(**params): + """按来源和页号构造结果，同时报告真实搜索观察字段。""" site, page = params["site"]["id"], params["page"] calls.append((site, params["keyword"], page)) failed = (site, page) in failures @@ -53,13 +62,120 @@ def manual_owner(monkeypatch, page_clock): async def events(iterator): + """消费完整事件流，供最终页结果和来源状态断言使用。""" return [item async for item in iterator] def site_facts(result): + """从最终状态中提取支持独立翻页的站点来源。""" return [item for item in result[-1]["sources"] if decode_source(item["source"])[0] != "plugin"] +@pytest.fixture +def blocked_site(manual_owner, monkeypatch): + """阻塞首个站点，其他站点即时完成，用同步信号验证并发和取消收口。""" + from app.runtime import tasks as tasks_module + + owner, _, _ = manual_owner + monkeypatch.setattr(owner, "_runtime_config", replace(owner.runtime_config, search_threadpool_size=2)) + monkeypatch.setattr(owner, "_runtime_config_provider", None) + registry = tasks_module.TaskRegistry() + monkeypatch.setattr(tasks_module, "_runtime_registry", registry) + started, release, finished = threading.Event(), threading.Event(), threading.Event() + original = owner.search_site_torrents + + def request(**params): + """让排序靠前的站点等待释放，证明快站点不受它阻塞。""" + if params["site"]["id"] != 1: + return original(**params) + started.set() + try: + assert release.wait(5), "测试未释放慢站点" + return original(**params) + finally: + finished.set() + + monkeypatch.setattr(owner, "search_site_torrents", request) + yield started, release, finished, registry + release.set() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("media_search", [False, True]) +@pytest.mark.parametrize("transport", [False, True]) +async def test_fast_site_preview_arrives_before_slow_site_and_final_page( + manual_owner, blocked_site, monkeypatch, media_search, transport, +): + """标题和精确搜索均先展示快站点，最终过滤前不提交来源页号。""" + from app.chain.media import MediaChain + + owner, calls, (_, saved) = manual_owner + started, release, finished, registry = blocked_site + params = {"keyword": "Show"} + if media_search: + media = MediaInfo(media_source=MediaSource.TMDB, media_id="1", tmdb_id=1, type=MediaType.TV, + title="Show", original_title="Show", names=["Show"]) + monkeypatch.setattr(MediaChain, "async_recognize_media", AsyncMock(return_value=media)) + monkeypatch.setattr(MediaChain, "async_supplement_media_info", AsyncMock(return_value=media)) + params = {"media_source": MediaSource.TMDB, "media_id": "1", "mtype": MediaType.TV} + stream = SearchManualOwner.events(owner, params=params) + if transport: + from app.api.endpoints import search as search_endpoint + + monkeypatch.setattr(search_endpoint, "_SSE_APPEND_FLUSH_INTERVAL", 0.01) + stream = search_endpoint._iter_batched_search_events(stream) + received = [] + try: + async with asyncio.timeout(3): + while True: + event = await anext(stream) + received.append(event) + if event.get("items"): + break + assert started.is_set() and not finished.is_set() + assert event["type"] == "append" and event["stage"] == "searching" + assert [item["torrent_info"]["site"] for item in event["items"]] == [2] + assert "sources" not in event + assert not any(item["type"] in {"replace", "done"} for item in received) + assert saved == [] + if media_search: + assert event["items"][0]["match_status"] == "candidate" + assert event["items"][0]["media_info_is_target"] is False + release.set() + received.extend(await events(stream)) + finally: + release.set() + await stream.aclose() + assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)] + assert [item["type"] for item in received[-2:]] == ["replace", "done"] + assert len(received[-2]["items"]) == 2 + assert [item["site_name"] for item in site_facts(received)] == ["A", "B"] + assert saved and not registry.records + + +@pytest.mark.asyncio +async def test_closing_preview_waits_for_inflight_workers_without_committing_page(manual_owner, blocked_site): + """离开搜索流时等待在途同步请求收尾，不写最终结果和翻页状态。""" + owner, _, (_, saved) = manual_owner + started, release, finished, registry = blocked_site + stream = SearchManualOwner.events(owner, params={"keyword": "Show"}) + try: + async with asyncio.timeout(3): + while not (await anext(stream)).get("items"): + pass + assert started.is_set() and not finished.is_set() + closing = asyncio.create_task(stream.aclose()) + await asyncio.sleep(0) + assert not closing.done() + release.set() + await asyncio.wait_for(closing, timeout=3) + assert finished.is_set() and not registry.records + assert saved == [] + finally: + release.set() + await stream.aclose() + + @pytest.mark.asyncio @pytest.mark.parametrize("requested,expected", [(None, None), (MediaType.UNKNOWN, None), (MediaType.TV, MediaType.TV), (MediaType.MOVIE, MediaType.MOVIE)]) @@ -75,10 +191,14 @@ async def test_title_search_keeps_unrestricted_mteam_categories_and_explicit_typ plugin_types = [] class FakeRequest: + """回放 M-Team 的分类与分页响应，不访问真实站点。""" + def __init__(self, **_params): + """接受原传输层参数以保持索引器调用协议。""" pass def post_res(self, _url=None, json=None, **_params): + """用分类和页号生成可重复的离线搜索响应。""" requests.append(json) count = 1 if json["categories"] == MTorrentSpider._movie_category else 3 offset = (json["pageNumber"] - 1) * count @@ -120,9 +240,10 @@ async def test_title_search_keeps_unrestricted_mteam_categories_and_explicit_typ @pytest.mark.asyncio async def test_first_request_loads_page_zero_of_every_source_and_saves_last_search(manual_owner): + """首页并发请求各来源，但完成页事实仍保持来源顺序。""" owner, calls, (_, saved) = manual_owner result = await events(SearchManualOwner.events(owner, params={"keyword": "Show"})) - assert calls == [(1, "Show", 0), (2, "Show", 0)] + assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)] sources = site_facts(result) assert [(item["site_name"], item["page"], item["can_continue"], item["error"]) for item in sources] == [ ("A", 0, True, None), ("B", 0, True, None)] @@ -149,6 +270,7 @@ async def test_continuation_restores_previous_page_and_keeps_last_search(manual_ @pytest.mark.asyncio @pytest.mark.parametrize("pageable", [True, False]) async def test_failed_source_keeps_its_page_for_retry_without_hiding_other_sources(manual_owner, monkeypatch, pageable): + """来源失败不影响其他来源，重试只请求原失败页。""" owner, calls, (failures, _) = manual_owner monkeypatch.setattr(owner, "get_search_page_size", lambda **_params: 100 if pageable else None) failures.add((1, 0)) @@ -156,11 +278,11 @@ async def test_failed_source_keeps_its_page_for_retry_without_hiding_other_sourc first, second = site_facts(result) assert (first["error"], first["page"], first["can_continue"]) == ("timeout", 0, True) assert (second["error"], second["page"], second["can_continue"]) == (None, 0, pageable) - assert calls == [(1, "Show", 0), (2, "Show", 0)] + assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)] failures.clear() retried = await events(SearchManualOwner.events(owner, params={ "keyword": "Show", "page": first["page"], "source": first["source"]})) - assert calls == [(1, "Show", 0), (2, "Show", 0), (1, "Show", 0)] + assert sorted(calls) == [(1, "Show", 0), (1, "Show", 0), (2, "Show", 0)] assert (site_facts(retried)[0]["error"], site_facts(retried)[0]["can_continue"]) == (None, pageable) @@ -179,13 +301,15 @@ async def test_alias_fallback_only_on_first_page_and_source_keeps_actual_keyword @pytest.mark.asyncio async def test_zero_display_rows_still_have_more(manual_owner, monkeypatch): + """标题预览和最终结果都为空时，原始页事实仍允许继续加载。""" from app.chain.search.manual import _ManualPage owner, _, _ = manual_owner # 过滤规则把本页结果全部过滤掉：显示为空，但站点仍有后续页。 - monkeypatch.setattr(_ManualPage, "results", lambda _self: []) + monkeypatch.setattr(_ManualPage, "results", lambda _self, _torrents=None: []) result = await events(SearchManualOwner.events(owner, params={"keyword": "Show", "sites": [1]})) assert result[-2]["items"] == [] + assert all(not item["items"] for item in result if item["type"] == "append") assert all(item["can_continue"] for item in site_facts(result)) @@ -231,6 +355,7 @@ async def test_site_without_pagination_has_only_the_first_page(manual_owner, mon @pytest.mark.asyncio @pytest.mark.parametrize("filter_all", [False, True]) async def test_raw_page_signature_stops_only_the_identical_adjacent_page(manual_owner, monkeypatch, filter_all): + """原始页摘要独立于展示过滤，仅相邻页完全重复时结束来源。""" from app.chain.search.manual import _ManualPage owner, calls, _ = manual_owner @@ -245,7 +370,7 @@ async def test_raw_page_signature_stops_only_the_identical_adjacent_page(manual_ for number in numbers] monkeypatch.setattr(owner, "search_site_torrents", request) if filter_all: - monkeypatch.setattr(_ManualPage, "results", lambda _self: []) + monkeypatch.setattr(_ManualPage, "results", lambda _self, _torrents=None: []) first = await events(SearchManualOwner.events(owner, params={"keyword": "Show", "sites": [1]})) first_source = site_facts(first)[0] assert calls == [0]