tests/test_downloads_resource_search.py
Rama master · réplica de lectura del repositorio original.
"""索引器搜索用合成 XML、假 HTTP 与临时凭据；不连接真实账号。"""
import hashlib
import json
import tempfile
import unittest
from pathlib import Path
from xml.sax.saxutils import escape
import httpx
from peach.follow_secrets import CredentialStore
from peach.resource_search import (CREDENTIAL, MAX_BYTES, MAX_TORRENTS, Filters, Indexers, Search, endpoint,
qualities)
HASH = "c9e15763f722f23e98a29decdfae341b98d53056"
CAPS = '<caps><limits max="25"/><searching><search available="yes" supportedParams="q"/></searching></caps>'
ROW = {"key": "one", "name": "来源一", "url": "http://indexer.test/api",
"api_key": "private-test-key", "enabled": True}
def item(title="ABC-123 4K HEVC 中字", info_hash=HASH, size=1024 ** 3, seeders=3,
magnet_field="magneturl"):
magnet = f"magnet:?xt=urn:btih:{info_hash}&tr=https://tracker.test/?passkey=secret"
link = (f'<torznab:attr name="magneturl" value="{escape(magnet, {chr(34): """})}"/>'
if magnet_field == "magneturl" else f'<{magnet_field}>{escape(magnet)}</{magnet_field}>')
return (f'<item><title>{escape(title)}</title><size>{size}</size>{link}'
f'<torznab:attr name="seeders" value="{seeders}"/>'
'<torznab:attr name="peers" value="7"/></item>')
#: 键按字节序排列的最小种子；infohash 是 info 字典原文的 SHA-1。
INFO = b"d6:lengthi1e4:name7:ABC-123e"
TORRENT = b"d8:announce16:http://t.test/an4:info" + INFO + b"e"
TORRENT_HASH = hashlib.sha1(INFO).hexdigest()
DOWNLOAD = "http://indexer.test/1/download?apikey=private-test-key&link=abc"
def torrent_item(title="ABC-123 FHD", link=DOWNLOAD, seeders=3):
"""OneJAV 一类条目：没有磁力与 infohash，只有索引器下载代理的种子地址。"""
return (f'<item><title>{escape(title)}</title><size>{1024 ** 3}</size><link>{link}</link>'
f'<torznab:attr name="seeders" value="{seeders}"/></item>')
def feed(*items):
return '<rss xmlns:torznab="http://torznab.com/schemas/2015/feed"><channel>' + ''.join(items) + '</channel></rss>'
class SearchTests(unittest.TestCase):
def test_edition_badges_follow_the_catalog_suffix_rules(self):
self.assertTrue(qualities("ABC-123-CH", [], "ABC-123")["chinese"])
self.assertTrue(qualities("ABC-123-UC", [], "ABC-123")["uncensored"])
self.assertFalse(qualities("ABC-123", [], "ABC-123")["chinese"])
def test_fhd_and_uhd_labels_count_as_resolutions(self):
self.assertEqual(qualities("[FHD] ABC-123", [], "ABC-123")["resolution"], 1080)
self.assertEqual(qualities("ABC-123 UHD", [], "ABC-123")["resolution"], 2160)
self.assertEqual(qualities("[HD Uncensored] ABC-123", [], "ABC-123")["resolution"], 0)
def search(self, content, *, rows=None, code="ABC-123", filters=None, blocked=()):
self.requests = []
def respond(request):
self.requests.append(request)
if callable(content):
return content(request)
return httpx.Response(200, text=CAPS if request.url.params["t"] == "caps" else content)
search = Search(transport=httpx.MockTransport(respond))
return search.run(rows or [ROW], code, filters or Filters(), blocked)
def test_caps_then_search_and_private_tracker_is_not_returned(self):
result = self.search(feed(item()))
self.assertTrue(result["ok"])
self.assertEqual([r.url.params["t"] for r in self.requests], ["caps", "search"])
params = self.requests[1].url.params
self.assertEqual((params["q"], params["cat"], params["limit"]), ("ABC-123", "6000", "25"))
row = result["items"][0]
self.assertEqual((row["info_hash"], row["resolution"], row["codec"], row["chinese"]),
(HASH, 2160, "HEVC", True))
self.assertNotIn("secret", json.dumps(result))
self.assertNotIn(ROW["api_key"], json.dumps(result))
def test_magnet_link_and_guid_are_accepted_by_content(self):
for field in ("magneturl", "link", "guid"):
with self.subTest(field=field):
self.assertEqual(self.search(feed(item(magnet_field=field)))["items"][0]["id"], HASH)
def torrent_search(self, *items, torrent=TORRENT, status=200, headers=None):
def respond(request):
if request.url.path.startswith("/1/download"):
return httpx.Response(status, content=torrent, headers=headers)
return httpx.Response(200, text=CAPS if request.url.params["t"] == "caps" else feed(*items))
return self.search(respond)
def test_torrent_only_item_becomes_a_magnet_from_its_info_dictionary(self):
result = self.torrent_search(torrent_item())
row = result["items"][0]
self.assertEqual((row["info_hash"], row["uri"], row["resolution"]),
(TORRENT_HASH, f"magnet:?xt=urn:btih:{TORRENT_HASH}", 1080))
self.assertEqual([r.url.path for r in self.requests], ["/api", "/api", "/1/download"])
self.assertNotIn(ROW["api_key"], json.dumps(result))
def test_proxy_redirect_to_a_magnet_is_read_from_location(self):
result = self.torrent_search(torrent_item(), status=301,
headers={"location": f"magnet:?xt=urn:btih:{HASH}"})
self.assertEqual(result["items"][0]["id"], HASH)
def test_torrent_link_off_the_indexer_origin_is_not_requested(self):
result = self.torrent_search(torrent_item(link="http://tracker.test/1/download/x.torrent"))
self.assertEqual(result["items"], [])
self.assertEqual({r.url.host for r in self.requests}, {"indexer.test"})
def test_unreadable_torrent_drops_only_that_item(self):
unsorted = b"d4:infod4:name1:a6:lengthi1eee"
for torrent in (b"<!DOCTYPE html>", unsorted, b"d8:announce1:xe"):
with self.subTest(torrent=torrent[:16]):
result = self.torrent_search(torrent_item(), item(), torrent=torrent)
self.assertTrue(result["ok"])
self.assertEqual([row["id"] for row in result["items"]], [HASH])
self.assertEqual(result["warnings"], [])
def test_torrent_fetches_skip_dead_items_and_stop_at_the_per_indexer_cap(self):
rows = [torrent_item(), torrent_item(seeders=0)] + [torrent_item() for _ in range(MAX_TORRENTS + 2)]
self.torrent_search(*rows)
self.assertEqual(sum(r.url.path.startswith("/1/download") for r in self.requests), MAX_TORRENTS)
def test_utf8_xml_with_bom_is_accepted(self):
def respond(request):
xml = CAPS if request.url.params["t"] == "caps" else feed(item())
return httpx.Response(200, content=("\ufeff" + xml).encode("utf-8"))
self.assertEqual(self.search(respond)["items"][0]["id"], HASH)
def test_fc2_query_uses_digits_and_keeps_exact_identity(self):
result = self.search(feed(item(title="FC2-PPV-1234567 1080p")), code="FC2-PPV-1234567")
self.assertEqual(self.requests[1].url.params["q"], "1234567")
self.assertEqual(len(result["items"]), 1)
def test_filters_reject_different_code_zero_seeders_size_and_blocked_hash(self):
result = self.search(feed(item(title="ABC-124"), item(seeders=0), item(size=1), item()),
filters=Filters(min_size=100), blocked=[HASH])
self.assertEqual(result["items"], [])
def test_when_nothing_passes_the_filters_the_reasons_are_counted(self):
result = self.search(feed(item(title="ABC-124"), item(title="ABC-125"), item(seeders=0), item(size=1)),
filters=Filters(min_size=100))
self.assertEqual(result["items"], [])
self.assertEqual(result["rejected"], {"code": 2, "seeders": 1, "size": 1})
self.assertEqual(result["warnings"],
["索引器返回了 4 条结果，都没通过筛选：番号不符 2 条、体积不在区间 1 条、没有人做种 1 条"])
def test_quality_goal_precedes_size_and_only_five_candidates_are_returned(self):
rows = [item(title=f"ABC-123 1080p {'中字' if at == 6 else ''}",
info_hash=f"{at:040x}", size=(at + 1) * 1024 ** 3) for at in range(7)]
result = self.search(feed(*rows), filters=Filters(goal="chinese"))
self.assertEqual(len(result["items"]), 5)
self.assertEqual(result["items"][0]["id"], f"{6:040x}")
def test_partial_failure_preserves_results_and_hides_remote_error(self):
def respond(request):
if request.url.host == "bad.test":
raise httpx.ConnectError(str(request.url), request=request)
return httpx.Response(200, text=CAPS if request.url.params["t"] == "caps" else feed(item()))
result = self.search(respond, rows=[ROW, {**ROW, "key": "bad", "url": "http://bad.test/api"}])
self.assertTrue(result["ok"])
self.assertEqual(len(result["items"]), 1)
self.assertEqual(len(result["warnings"]), 1)
self.assertNotIn(ROW["api_key"], json.dumps(result))
def test_repeated_hash_merges_origins(self):
result = self.search(feed(item()), rows=[ROW, {**ROW, "key": "two", "name": "来源二"}])
self.assertEqual(len(result["items"]), 1)
self.assertEqual(set(result["items"][0]["origins"]), {"来源一", "来源二"})
def test_unavailable_search_does_not_send_search_request(self):
result = self.search(lambda request: httpx.Response(200, text='<caps><searching/></caps>'))
self.assertFalse(result["ok"])
self.assertEqual(len(self.requests), 1)
def test_bad_xml_entities_large_response_and_html_fail_without_results(self):
for content in ('<!DOCTYPE rss [<!ENTITY x "x">]><rss/>', '<html/>', '<rss>', 'x' * (MAX_BYTES + 1)):
with self.subTest(content=content[:30]):
result = self.search(content)
self.assertFalse(result["ok"])
self.assertEqual(result["items"], [])
def test_redirect_is_not_followed(self):
result = self.search(lambda request: httpx.Response(302, headers={"location": "http://other.test/"}))
self.assertFalse(result["ok"])
self.assertEqual(len(self.requests), 1)
def test_cache_rechecks_blacklist_and_expires(self):
calls = []
clock = [0]
def respond(request):
calls.append(request)
return httpx.Response(200, text=CAPS if request.url.params["t"] == "caps" else feed(item()))
search = Search(transport=httpx.MockTransport(respond), clock=lambda: clock[0])
self.assertEqual(len(search.run([ROW], "ABC-123", Filters())["items"]), 1)
self.assertEqual(search.run([ROW], "ABC-123", Filters(), [HASH])["items"], [])
self.assertEqual(len(calls), 2)
clock[0] = 61
search.run([ROW], "ABC-123", Filters())
self.assertEqual(len(calls), 4)
def test_disabled_and_busy_search_do_not_contact_indexer(self):
search = Search(transport=httpx.MockTransport(lambda request: self.fail("unexpected request")))
self.assertEqual(search.run([{**ROW, "enabled": False}], "ABC-123", Filters())["state"], "unavailable")
with search.lock:
self.assertEqual(search.run([ROW], "ABC-123", Filters())["state"], "busy")
def test_invalid_code_and_size_range_are_rejected(self):
for code in ("", "not-a-code", "x" * 90):
with self.assertRaises(ValueError):
Search().run([ROW], code, Filters())
for values in ({"min_size": -1}, {"min_size": 10, "max_size": 2}, {"goal": "unknown"}):
with self.assertRaises(ValueError):
Filters(**values)
class SettingsTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.store = CredentialStore(Path(temporary.name))
self.settings = Indexers(self.store)
def test_save_preserve_clear_disable_and_remove_are_reversible(self):
public = self.settings.save({"indexers": [ROW]})
self.assertNotIn(ROW["api_key"], json.dumps(public))
self.assertTrue(public["indexers"][0]["api_key_set"])
row = {**ROW, "api_key": "", "enabled": False}
self.settings.save({"indexers": [row]})
self.assertEqual(self.settings.load()[0]["api_key"], ROW["api_key"])
self.assertFalse(self.settings.load()[0]["enabled"])
self.settings.save({"indexers": [{**row, "clear_api_key": True}]})
self.assertFalse(self.settings.public()["indexers"][0]["api_key_set"])
self.settings.save({"indexers": []})
self.assertEqual(self.settings.load(), [])
def test_changing_address_does_not_forward_stored_secret(self):
self.settings.save({"indexers": [ROW]})
self.settings.save({"indexers": [{**ROW, "url": "http://other.test/api", "api_key": ""}]})
self.assertEqual(self.settings.load()[0]["api_key"], "")
def test_validation_is_atomic_and_rejects_duplicate_ids_and_secret_urls(self):
self.settings.save({"indexers": [ROW]})
for rows in ([ROW, ROW], [{**ROW, "url": "http://indexer.test/api?apikey=secret"}], [ROW] * 5):
with self.assertRaises(ValueError):
self.settings.save({"indexers": rows})
self.assertEqual(self.settings.load(), [ROW])
for value in ("file:///tmp/x", "http://user:pass@host/api", "https://host/api#secret"):
with self.assertRaises(ValueError):
endpoint(value)
def test_corrupt_configuration_does_not_disclose_file_contents(self):
self.store.save(CREDENTIAL, {"indexers": 'private-test-key'})
with self.assertRaisesRegex(ValueError, "配置读不出来") as error:
self.settings.public()
self.assertNotIn(ROW["api_key"], str(error.exception))