"""站点解析器契约：取页与解析分开、配置注入、`query` 的异常翻译、失败原因表，以及配置与各张表的一致。"""
import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import patch

import httpx

from peach import scraping_access
from peach.http import HttpResponse
from peach.jav_cover_fetch import DeadlineExceeded, HostLimitedTransport, NotFound, Unavailable
from peach.library_processing import PROVIDER_NAMES, SOURCE_INTERVALS, SOURCE_LABELS
from peach.metadata import MetadataProviderError
from peach.metadata_policy import SOURCE_SPECS
from peach.metadata_routes import FC2_STAGE
from peach.scraping_access import FIRST_BLOCKED_PAUSE, SourcePaused, SourceTransport, cooldown_state
from peach.sources import (COOLDOWN_ACTIONS, PERMANENT_REASONS, REASON_KINDS, SEESAA, SITE_SOURCES, FailureReason,
                           Page, Session, SiteConfig, SiteRecord, SiteSource, SourceFailure, http_failure)
from peach.sources.base import challenge_page
from peach.sources.seesaa import WIKI_SOURCES, WikiPages

DEMO = SiteConfig(name="demo", label="Demo", provider="demo-page", base_url="https://demo.test",
                  domains=("demo.test",), stage="community")


class JavDBResourceTests(unittest.TestCase):
    def test_title_projection_accepts_original_only_display_only_and_english_original(self):
        from peach.sources.javdb import JavDBSource
        panel = '<div class="panel-block"><strong>番號:</strong><span class="value">MIKR-023</span></div>'
        for html, expected in (('<span class="origin-title">日本語原題</span>', '日本語原題'),
                               ('<strong class="current-title">仅显示标题</strong>', '仅显示标题'),
                               ('<strong class="current-title">译文</strong><span class="origin-title">Original *Title*</span>',
                                'Original *Title*')):
            with self.subTest(html=html):
                payload = JavDBSource().parse(Page('https://javdb.com/v/XeYkdb', (html + panel).encode()), 'MIKR-023').payload()
                self.assertEqual(payload['title'], expected)

    def test_original_title_is_separate_from_localized_display_and_keeps_source_identity(self):
        from peach.sources.javdb import JavDBSource
        html = '''<strong class="current-title">自动翻译标题</strong>
        <span class="origin-title" style="display:none">Original *A* **j*yqy*pq* 日本語</span>
        <div class="panel-block"><strong>番號:</strong><span class="value">MIKR-023</span></div>'''
        payload = JavDBSource().parse(Page('https://javdb.com/v/XeYkdb', html.encode()), 'MIKR-023').payload()
        self.assertEqual(payload['title'], 'Original *A* **j*yqy*pq* 日本語')
        self.assertEqual(payload['origin_title'], payload['title'])
        self.assertEqual(payload['display_title'], '自动翻译标题')
        self.assertEqual(payload['source_id'], 'XeYkdb')
        self.assertEqual(payload['javdb_rule'], 2)

    def test_same_number_different_source_ids_are_not_selected_by_search_order(self):
        from peach.sources.javdb import JavDBSource
        source = JavDBSource()
        for identities in [('XeYkdb', 'MO50k'), ('MO50k', 'XeYkdb')]:
            html = ''.join(f'<a href="/v/{identity}" class="box" title="film"><strong>MIKR-023</strong></a>'
                           for identity in identities)
            with self.subTest(order=identities), patch.object(source, '_page', return_value=Page('https://javdb.com/search', html.encode())) as get:
                with self.assertRaises(SourceFailure) as caught:
                    source.fetch('MIKR-023', session=None)
                self.assertEqual(caught.exception.detail, 'identity_ambiguous')
                self.assertEqual(get.call_count, 1)

    def test_repeated_card_for_one_source_id_is_not_a_second_work(self):
        from peach.sources.javdb import JavDBSource
        source = JavDBSource()
        card = '<a href="/v/XeYkdb" class="box" title="film"><strong>MIKR-023</strong></a>'
        detail = Page('https://javdb.com/v/XeYkdb', b'detail')
        with patch.object(source, '_page', side_effect=[Page('https://javdb.com/search', (card * 2).encode()), detail]):
            self.assertEqual(source.fetch('MIKR-023', session=None).url, detail.url)

    def test_categories_keep_raw_words_and_deduplicate_links_after_identity_check(self):
        from peach.sources.javdb import JavDBSource
        from peach.metadata import extract_peach_fields
        html = '''<div class="panel-block"><strong>番號:</strong><span class="value">ABW-220</span></div>
        <div class="panel-block"><strong>類別:</strong><span class="value">
        <a href="/tags?c=1">高跟鞋</a><a href="/tags?c=2">絲襪、過膝襪</a>
        <a href="/tags?c=3"><b>尚未收錄的分類</b></a><a href="/tags?c=1">高跟鞋</a>
        <a href="/tags?c=4">單體作品</a></span></div>'''
        page = Page("https://javdb.com/v/demo", html.encode())
        record = JavDBSource().parse(page, "ABW-220")
        self.assertEqual(record.tags, ("高跟鞋", "絲襪、過膝襪", "尚未收錄的分類", "單體作品"))
        self.assertEqual(record.payload()["genres"], ["高跟鞋", "絲襪、過膝襪", "尚未收錄的分類", "單體作品"])
        fields = extract_peach_fields(record.payload())
        self.assertEqual(fields["tags"]["value"], ["高跟", "丝袜"])
        self.assertEqual(fields["tags"]["unmapped_genres"], ["尚未收錄的分類"])
        with self.assertRaises(SourceFailure):
            JavDBSource().parse(page, "ABW-221")

    def test_resource_area_and_comment_links_share_identity(self):
        from peach.sources.javdb import parse_comment_text, merge_resources
        digest = 'a' * 40
        file_hash = 'b' * 32
        magnet = {'id': 'magnet:' + digest, 'protocol': 'magnet', 'info_hash': digest,
                  'uri': f'magnet:?xt=urn:btih:{digest}', 'name': '演示', 'size': '4.00GB', 'files': '2',
                  'date': '2026-10-01', 'attributes': ['中字'], 'source': 'JavDB',
                  'source_url': 'https://javdb.com/v/demo', 'origins': ['JavDB']}
        text = (f'magnet:?xt=urn:btih:{digest.upper()} ed2k://|file|another-name|123456|{file_hash}|/\n'
                f'ed2k://|file|demo%20file.mkv|123456|{file_hash.upper()}|/\n'
                'https://EXAMPLE.test/file?q=1 https://example.test/file?q=1 javascript:alert(1)')
        comments = parse_comment_text(text, 'https://javdb.com/v/demo#review-item-1', '2026-10-02')
        items = merge_resources([magnet, *comments])
        self.assertEqual([item['protocol'] for item in items], ['magnet', 'ed2k', 'url'])
        self.assertEqual(items[0]['size'], '4.00GB')
        self.assertEqual(items[0]['attributes'], ['中字'])
        self.assertEqual(len(items[0]['origins']), 2)
        self.assertEqual(items[1]['name'], 'another-name')
        self.assertEqual(items[1]['size'], '123,456 B')
        self.assertEqual(items[2]['uri'], 'https://example.test/file?q=1')


class DemoSource(SiteSource):
    """最小的一站：作品页是 `/<番号>`，页面正文就是标题。"""

    DEFAULT = DEMO

    def fetch(self, code, *, session):
        return session.get(f"{self.config.base_url}/{code}", config=self.config)

    def parse(self, page, code):
        if page.text == "gate":
            raise SourceFailure(FailureReason.AUTH_REQUIRED, "Demo 要求登录")
        return SiteRecord(source=self.config.name, provenance=self.config.provider, code=code,
                          source_url=page.url, title=page.text)


def serve(pages):
    calls = []

    def call(request, timeout, limit):
        calls.append((request.url, request.headers.get("Referer"), limit))
        body = pages.get(request.url)
        if isinstance(body, Exception):
            raise body
        return HttpResponse(200 if body is not None else 404, {}, body or b"", request.url)

    call.calls = calls
    return call


class ContractTests(unittest.TestCase):
    def test_parse_reads_a_page_without_any_transport(self):
        record = DemoSource().parse(Page("https://demo.test/ABC-001", "標題".encode()), "ABC-001")
        self.assertEqual((record.source, record.provenance, record.title, record.source_url),
                         ("demo", "demo-page", "標題", "https://demo.test/ABC-001"))

    def test_fetch_uses_the_config_for_referer_and_page_limit_and_query_chains_both_steps(self):
        transport = serve({"https://demo.test/ABC-001": "標題".encode()})
        record = DemoSource().query("ABC-001", session=Session(transport))
        self.assertEqual(record.title, "標題")
        self.assertEqual(transport.calls, [("https://demo.test/ABC-001", "https://demo.test/", DEMO.page_limit)])

    def test_an_injected_config_replaces_the_default_one(self):
        mirror = SiteConfig(name="demo", label="Demo", provider="demo-page", base_url="https://mirror.test",
                            domains=("mirror.test",), stage="community", page_limit=1024)
        transport = serve({"https://mirror.test/ABC-001": b"x"})
        DemoSource(mirror).query("ABC-001", session=Session(transport))
        self.assertEqual(transport.calls, [("https://mirror.test/ABC-001", "https://mirror.test/", 1024)])
        self.assertIs(DemoSource().config, DEMO)

    def test_one_request_can_name_its_own_referer_and_extra_headers(self):
        seen = []

        def transport(request, timeout, limit):
            seen.append((request.headers.get("Referer"), request.headers.get("X-Inertia")))
            return HttpResponse(200, {}, b"x", request.url)

        session = Session(transport)
        session.get("https://demo.test/ABC-001", config=DEMO)
        session.get("https://demo.test/ABC-001", config=DEMO, referer="https://demo.test/ABC-001",
                    headers={"X-Inertia": "true"})
        self.assertEqual(seen, [("https://demo.test/", None), ("https://demo.test/ABC-001", "true")])

    def test_post_sends_the_body_with_the_same_referer_limit_and_extra_headers(self):
        """只收 POST 的接口（DMM 的 GraphQL）走 `post`：方法与请求体进 `HttpRequest`，其余与 `get` 同一条路。"""
        seen = []

        def transport(request, timeout, limit):
            seen.append((request.method, request.url, request.body, request.headers.get("Referer"),
                         request.headers.get("Content-Type"), limit))
            return HttpResponse(200, {}, b'{"data": {}}', request.url)

        page = Session(transport).post("https://demo.test/graphql", config=DEMO, body=b'{"query": "x"}',
                                       headers={"Content-Type": "application/json"})
        self.assertEqual(seen, [("POST", "https://demo.test/graphql", b'{"query": "x"}', "https://demo.test/",
                                 "application/json", DEMO.page_limit)])
        self.assertEqual((page.url, page.body), ("https://demo.test/graphql", b'{"data": {}}'))
        Session(transport).get("https://demo.test/ABC-001", config=DEMO)
        self.assertEqual(seen[-1][:3], ("GET", "https://demo.test/ABC-001", None))

    def test_records_default_to_the_single_query_result(self):
        transport = serve({"https://demo.test/ABC-001": "標題".encode()})
        found = DemoSource().records("ABC-001", session=Session(transport))
        self.assertEqual([record.title for record in found], ["標題"])
        with self.assertRaises(SourceFailure):
            DemoSource().records("ABC-002", session=Session(transport))

    def test_query_translates_http_tiers_into_reasons_and_keeps_the_wording(self):
        with self.assertRaises(SourceFailure) as caught:
            DemoSource().query("ABC-001", session=Session(serve({})))
        self.assertEqual((caught.exception.reason, caught.exception.status_code, str(caught.exception)),
                         (FailureReason.NOT_FOUND, 404, "HTTP 404"))
        for status, reason in ((401, FailureReason.AUTH_REQUIRED), (403, FailureReason.AUTH_REQUIRED),
                               (410, FailureReason.GONE), (429, FailureReason.RATE_LIMITED),
                               (500, FailureReason.SERVER_ERROR), (503, FailureReason.SERVER_ERROR),
                               (418, FailureReason.NETWORK)):
            with self.subTest(status=status):
                failure = http_failure(Unavailable(f"HTTP {status}"))
                self.assertEqual((failure.reason, failure.status_code, failure.message), (reason, status, f"HTTP {status}"))
        self.assertEqual(http_failure(NotFound("来源给的是占位图")).reason, FailureReason.NOT_FOUND)

    def test_query_passes_parse_failures_through_and_leaves_transport_signals_alone(self):
        """冷却、预算与连接失败由传输层抛、由调用方按原语义处理，契约不改写它们。"""
        with self.assertRaises(SourceFailure) as caught:
            DemoSource().query("ABC-001", session=Session(serve({"https://demo.test/ABC-001": b"gate"})))
        self.assertEqual(caught.exception.reason, FailureReason.AUTH_REQUIRED)
        for signal in (SourcePaused("来源正在冷却"), DeadlineExceeded("动作预算已用尽"),
                       httpx.TransportError("来源连接未取得")):
            with self.subTest(signal=type(signal).__name__), \
                    patch("peach.jav_cover_fetch.NETWORK_RETRY_DELAYS", ()), \
                    self.assertRaises(type(signal)):
                DemoSource().query("ABC-001", session=Session(serve({"https://demo.test/ABC-001": signal})))

    def test_the_record_projects_into_the_snapshot_shaped_payload(self):
        record = SiteRecord(source="demo", provenance="demo-page", code="ABC-001", source_url="https://demo.test/ABC-001",
                            title="標題", performers=({"japanese_name": "女優"},), studio="片商", runtime=90,
                            cover_urls=("https://demo.test/a.jpg", "https://demo.test/b.jpg"))
        self.assertEqual(record.cover_url, "https://demo.test/a.jpg")
        self.assertEqual(record.payload(), {
            "id": "ABC-001", "source_url": "https://demo.test/ABC-001", "title": "標題",
            "actresses": [{"japanese_name": "女優"}], "maker": "片商", "label": "", "series": "", "director": "",
            "release_date": "", "runtime": 90, "cover_urls": ["https://demo.test/a.jpg", "https://demo.test/b.jpg"],
            "cover_url": "https://demo.test/a.jpg"})
        tagged = SiteRecord(source="demo", provenance="demo-page", code="ABC-001", source_url="", tags=(),
                            extra={"plot": "剧情"})
        payload = tagged.payload()
        self.assertEqual((payload["genres"], payload["plot"], payload["cover_urls"], payload["cover_url"]),
                         ([], "剧情", [], ""))
        self.assertNotIn("genres", record.payload(), "不给标签的站不带 genres 键")


class FailureReasonTests(unittest.TestCase):
    def test_the_table_has_twelve_reasons_each_with_a_kind(self):
        self.assertEqual({reason.value for reason in FailureReason}, {
            "cloudflare_challenge", "ip_banned", "geo_restricted", "auth_required", "not_found", "gone",
            "parse_error", "no_usable_metadata", "rate_limited", "timeout", "server_error", "network"})
        self.assertEqual(set(REASON_KINDS), set(FailureReason))
        self.assertEqual(set(REASON_KINDS.values()), {"auth", "unavailable", "not_found"})
        self.assertEqual({reason for reason, kind in REASON_KINDS.items() if kind == "auth"},
                         {FailureReason.CLOUDFLARE_CHALLENGE, FailureReason.IP_BANNED,
                          FailureReason.GEO_RESTRICTED, FailureReason.AUTH_REQUIRED})
        self.assertEqual({reason for reason, kind in REASON_KINDS.items() if kind == "not_found"},
                         {FailureReason.NOT_FOUND, FailureReason.GONE, FailureReason.NO_USABLE_METADATA})

    def test_provider_errors_keep_the_three_tier_semantics(self):
        auth = SourceFailure(FailureReason.CLOUDFLARE_CHALLENGE, "撞上 Cloudflare 挑战页", status_code=403)
        error = auth.provider_error("AVSOX")
        self.assertIsInstance(error, MetadataProviderError)
        self.assertEqual((error.kind, error.status_code, error.retryable, error.temporary, error.detail),
                         ("auth", 403, False, True, "cloudflare_challenge"))
        self.assertEqual(str(error), "AVSOX 需要登录或已被拒绝：撞上 Cloudflare 挑战页")
        missing = SourceFailure(FailureReason.NOT_FOUND, "javdb 没有这个番号").provider_error("javdb")
        self.assertEqual((missing.kind, str(missing)), ("not_found", "javdb 没有这个番号"))
        parse = SourceFailure(FailureReason.PARSE_ERROR, "结构对不上").provider_error("javdb")
        self.assertEqual((parse.kind, parse.retryable, parse.temporary), ("unavailable", False, False))
        timeout = SourceFailure(FailureReason.TIMEOUT, "超时", detail="curl 28").provider_error("javdb")
        self.assertEqual((timeout.kind, timeout.retryable, timeout.temporary, timeout.detail),
                         ("unavailable", True, True, "curl 28"))
        self.assertEqual(PERMANENT_REASONS, {FailureReason.PARSE_ERROR})

    def test_only_bans_and_rate_limits_pause_a_whole_site(self):
        self.assertEqual(COOLDOWN_ACTIONS, {FailureReason.CLOUDFLARE_CHALLENGE: "blocked",
                                            FailureReason.IP_BANNED: "blocked",
                                            FailureReason.GEO_RESTRICTED: "blocked",
                                            FailureReason.RATE_LIMITED: "rate_limited"})
        for reason in FailureReason:
            with self.subTest(reason=reason):
                self.assertEqual(SourceFailure(reason, "").cooldown_action, COOLDOWN_ACTIONS.get(reason, ""))
        self.assertEqual(set(COOLDOWN_ACTIONS.values()), {"blocked", "rate_limited"},
                         "两档就是 scraping_access.pause_source 的 refused=True / retry_after 两档")


CHALLENGE_HTML = (b"<!DOCTYPE html><html><head><title>Just a moment...</title></head>"
                  b"<body><script>window._cf_chl_opt={}</script></body></html>")


class SiteCooldownTests(unittest.TestCase):
    """自写站报出 `COOLDOWN_ACTIONS` 里的细档时整站进冷却，与 amane 桥写同一份记录，同一轮里不再一部片撞一次。"""

    def setUp(self):
        temporary = tempfile.TemporaryDirectory()
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name)
        finder = patch("peach.browser_transport.find_browser", return_value=None)
        finder.start()
        self.addCleanup(finder.stop)

    def session(self, source, respond):
        """真的 `SourceTransport` 套在 `HostLimitedTransport` 里（与 `LibraryMetadataProvider` 同一个形状），底下是假传输。"""
        calls = []

        def fake(request, timeout, limit):
            calls.append(request.url)
            return respond(request)

        raw = SourceTransport(self.root)
        raw.transports[source] = fake
        return Session(HostLimitedTransport(raw, 0.0)), calls

    def test_an_avbase_challenge_served_with_200_pauses_the_whole_site(self):
        session, calls = self.session("avbase", lambda request: HttpResponse(200, {}, CHALLENGE_HTML, request.url))
        with self.assertRaises(SourceFailure) as caught:
            SITE_SOURCES["avbase"]().query("ABW-358", session=session)
        self.assertEqual(caught.exception.reason, FailureReason.CLOUDFLARE_CHALLENGE)
        until, blocks = cooldown_state(self.root, "avbase")
        self.assertEqual((round(until - time.time()), blocks), (FIRST_BLOCKED_PAUSE, 1))
        with self.assertRaises(SourcePaused):
            SITE_SOURCES["avbase"]().query("MIDE-594", session=session)
        self.assertEqual(len(calls), 1, "冷却期内第二部片一次都不问")

    def test_the_challenge_title_is_recognised_in_its_other_wordings(self):
        for title in ("Just a moment...", "请稍候…", "Attention Required! | Cloudflare"):
            with self.subTest(title=title):
                self.assertTrue(challenge_page(f"<html><head><title> {title} </title>".encode()))
        self.assertFalse(challenge_page(b"<html><head><title>ABW-358 - AVBase</title>"
                                        b"<script src='/cdn-cgi/challenge-platform/scripts/jsd/main.js'>"),
                         "正常页里 Cloudflare 注入的脚本不算验证页")

    def test_each_cooldown_reason_goes_to_its_own_tier_and_other_reasons_leave_no_record(self):
        class Failing(DemoSource):
            reason = FailureReason.NOT_FOUND

            def parse(self, page, code):
                raise SourceFailure(self.reason, "demo")

        config = SiteConfig(name="javbus", label="JavBus", provider="javbus-page", base_url="https://www.javbus.com",
                            domains=("javbus.com",), stage="community")
        for reason in FailureReason:
            with self.subTest(reason=reason):
                scraping_access.cooldown_path(self.root, "javbus").unlink(missing_ok=True)
                session, _calls = self.session("javbus", lambda request: HttpResponse(200, {}, b"page", request.url))
                Failing.reason = reason
                with self.assertRaises(SourceFailure):
                    Failing(config).query("ABC-001", session=session)
                until, blocks = cooldown_state(self.root, "javbus")
                action = COOLDOWN_ACTIONS.get(reason, "")
                if action == "blocked":
                    self.assertEqual((round(until - time.time()), blocks), (FIRST_BLOCKED_PAUSE, 1))
                elif action == "rate_limited":
                    self.assertEqual((round(until - time.time()), blocks), (900, 0))
                else:
                    self.assertEqual((until, blocks), (0.0, 0), "不是整站的问题就不停整站")

    def test_a_nested_query_counts_one_failure_once(self):
        """子类的 `query` 包着基类的 `query`（`r18dev`），同一个失败只翻一次倍。"""
        class Nested(DemoSource):
            def parse(self, page, code):
                raise SourceFailure(FailureReason.IP_BANNED, "demo")

            def query(self, code, *, session):
                with self.holding(session):
                    return super().query(code, session=session)

        config = SiteConfig(name="javdb", label="javdb", provider="javdb-page", base_url="https://javdb.com",
                            domains=("javdb.com",), stage="community")
        session, _calls = self.session("javdb", lambda request: HttpResponse(200, {}, b"page", request.url))
        with self.assertRaises(SourceFailure):
            Nested(config).query("ABC-001", session=session)
        self.assertEqual(cooldown_state(self.root, "javdb")[1], 1)

    def test_a_javbus_refusal_pauses_the_site_like_the_other_community_sources(self):
        session, calls = self.session("javbus", lambda request: HttpResponse(403, {}, b"", request.url))
        for code in ("MIDE-594", "ABW-358"):
            with self.assertRaises(SourcePaused):
                SITE_SOURCES["javbus"]().query(code, session=session)
        self.assertEqual(len(calls), 1)
        self.assertEqual(cooldown_state(self.root, "javbus")[1], 1)


class ConfigConsistencyTests(unittest.TestCase):
    """配置是数据：`SOURCE_SPECS`、`SOURCE_LABELS`、`PROVIDER_NAMES`、`scraping_access.SOURCES` 与
    `SOURCE_INTERVALS` 里这几站的那几行，与站的配置逐项一致。"""

    def test_every_registered_site_agrees_with_the_existing_tables(self):
        for name, site in SITE_SOURCES.items():
            config = site.DEFAULT
            with self.subTest(site=name):
                self.assertIs(site().config, config)
                self.assertEqual(config.name, name)
                self.assertEqual(config.stage, SOURCE_SPECS[name].kind)
                self.assertEqual(config.label, SOURCE_LABELS[name])
                self.assertEqual(config.provider, PROVIDER_NAMES[name])
                if name in WIKI_SOURCES:
                    continue
                access = scraping_access.SOURCES[name]
                self.assertEqual(set(config.domains), set(access["domains"]))
                self.assertEqual(config.cookie, bool(access.get("cookie")))
                self.assertTrue(config.base_url.startswith("https://"))
                self.assertEqual(scraping_access.source_for(config.base_url), name)
                for host, interval in SOURCE_INTERVALS.items():
                    if scraping_access.source_for("https://" + host + "/") == name:
                        self.assertEqual(interval, config.interval, f"{host} 的主机间隔与配置不一致")

    def test_every_bridge_site_agrees_with_the_same_tables(self):
        """经 amane 桥的站同样逐项一致；每个站只有一个归属，桥站与自写站不重名（ADR-0048）。"""
        from peach import metadata_amane
        self.assertEqual(set(metadata_amane.SITE_CONFIGS) & set(SITE_SOURCES), set())
        for name, config in metadata_amane.SITE_CONFIGS.items():
            with self.subTest(site=name):
                self.assertEqual(config.name, name)
                self.assertEqual(SOURCE_SPECS[name].kind, "community" if config.stage == "amane" else config.stage)
                self.assertEqual(config.label, SOURCE_LABELS[name])
                self.assertEqual(config.provider, PROVIDER_NAMES[name])
                # Prestige 与 MGStage 的冷却记录和封面那一路同键：同一个出口对同一站只有一份冷却。
                if name in scraping_access.SOURCES:
                    self.assertEqual(scraping_access.SOURCES[name]["label"], config.label)
        self.assertEqual({name for name, config in metadata_amane.SITE_CONFIGS.items() if config.stage == "official"},
                         {"makers", "prestige", "faleno", "dahlia", "mgstage"})

    def test_seesaa_brings_its_own_transport_and_keeps_its_source_identity(self):
        """Seesaa 作品表不收 Cookie、采集设置页没有它的卡片，冷却只在本批内，取页、缓存与限额在 `WikiPages`，
        间隔与页面上限取自配置；provenance 是账本里已有的 `sougouwiki`。`scrape_codes` 点名时它自带传输，
        素人链上则借资料处理那一份（`LibraryMetadataProvider.wiki`）。"""
        self.assertEqual(set(WIKI_SOURCES), {"sougouwiki", "av_neme", "av_name"})
        for name, site in WIKI_SOURCES.items():
            config = site.DEFAULT
            with self.subTest(site=name):
                self.assertNotIn(name, scraping_access.SOURCES)
                self.assertIsNone(scraping_access.source_for(config.base_url))
                self.assertEqual((config.provider, config.cookie, config.interval, config.page_limit),
                                 (name, False, 2.0, 4 * 1024 * 1024))
                self.assertTrue(site.root.startswith(config.base_url + "/"))
        with patch("peach.sources.seesaa.HostLimiter") as limiter:
            pages = WikiPages("unused", transport=object())
        limiter.assert_called_once_with({}, default_interval=SEESAA.interval)
        self.assertIs(pages.config, SEESAA)

    def test_javdb_keeps_the_user_set_interval_and_the_other_sites_the_default(self):
        self.assertEqual(SITE_SOURCES["javdb"].DEFAULT.interval, 3.0)
        for name in ("javbus", "avbase", "r18dev", "1pondo", "fc2", "fc2cmadb", "fc2ppvdb", "javten", "javarchive",
                     *WIKI_SOURCES):
            self.assertEqual(SITE_SOURCES[name].DEFAULT.interval, 2.0, name)
        self.assertEqual({host for host in SOURCE_INTERVALS if scraping_access.source_for("https://" + host + "/") == "javdb"},
                         set(SOURCE_INTERVALS), "SOURCE_INTERVALS 里只有 javdb 的主机单独设间隔")

    def test_the_fourteen_sites_are_registered_with_their_stage_cookie_and_page_limit(self):
        shape = {name: (site.DEFAULT.stage, site.DEFAULT.cookie, site.DEFAULT.page_limit)
                 for name, site in SITE_SOURCES.items()}
        self.assertEqual(shape, {"r18dev": ("official_mirror", False, 2 * 1024 * 1024),
                                 "dmm": ("official", False, 1024 * 1024),
                                 "1pondo": ("official", False, 1024 * 1024),
                                 "fc2": ("official", False, 2 * 1024 * 1024),
                                 "fc2cmadb": ("community", True, 2 * 1024 * 1024),
                                 "fc2ppvdb": ("community", True, 2 * 1024 * 1024),
                                 "javten": ("community", True, 2 * 1024 * 1024),
                                 "javarchive": ("community", False, 2 * 1024 * 1024),
                                 "avbase": ("community", False, 4 * 1024 * 1024),
                                 "javbus": ("community", True, 4 * 1024 * 1024),
                                 "javdb": ("community", True, 4 * 1024 * 1024),
                                 "sougouwiki": ("community", False, 4 * 1024 * 1024),
                                 "av_neme": ("community", False, 4 * 1024 * 1024),
                                 "av_name": ("community", False, 4 * 1024 * 1024)})
        self.assertEqual({name for name in SITE_SOURCES if SOURCE_SPECS[name].official},
                         {"r18dev", "dmm", "1pondo", "fc2"})
        self.assertEqual(tuple(name for name in SITE_SOURCES if name in FC2_STAGE), FC2_STAGE,
                         "FC2 五站按链上先后登记")
        # 在 Cloudflare 验证后面的两站：公开采集带用户贴的 Cookie，403 时整站冷却。
        for name in ("fc2ppvdb", "javten"):
            self.assertEqual({key: scraping_access.SOURCES[name].get(key) for key in ("cookie", "session")},
                             {"cookie": True, "session": True}, name)
            self.assertTrue(scraping_access.SOURCES[name]["blocked_pause"], name)


if __name__ == "__main__":
    unittest.main()