import threading
import json
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

import httpx

from peach import http as peach_http
from peach.follow_stream import (
    MAX_PROXY_REDIRECTS, FollowMediaResolver, FollowMediaUnavailable,
    FollowProxyError, ResolvedFollowMedia, open_upstream, proxy_request_headers,
    proxyable,
)
from peach.follow_secrets import Credential
from peach.http import HttpResponse


class FollowMediaResolverTests(unittest.TestCase):
    def test_source_cooldown_survives_resolver_restart_and_expires(self):
        moment, wall = [1000.0], [11000.0]
        transport = mock.Mock()
        with tempfile.TemporaryDirectory() as temporary:
            options = {'state_root': Path(temporary), 'clock': lambda: moment[0],
                       'wall_clock': lambda: wall[0]}
            first = FollowMediaResolver(transport, **options)
            first.pause('iwara', 120)
            wall[0] += 30
            moment[0] = 5
            second = FollowMediaResolver(transport, **options)
            self.assertEqual(second.cooling('iwara'), 90)
            moment[0] += 91
            wall[0] += 91
            self.assertEqual(FollowMediaResolver(transport, **options).cooling('iwara'), 0)
            transport.assert_not_called()

    def test_iwara_transport_cleanup_preserves_the_callers_transport(self):
        owned = mock.Mock()
        upstream = peach_http.HttpxTransport()
        with mock.patch("peach.follow_stream.IwaraConnector", return_value=SimpleNamespace(transport=owned)):
            resolver = FollowMediaResolver(upstream)
        resolver.close()
        owned.close.assert_called_once_with()
        upstream.close()
        injected = mock.Mock()
        FollowMediaResolver(injected).close()
        injected.close.assert_not_called()

    def test_iwara_playback_uses_shared_cache_quality_selection_and_no_media_credentials(self):
        requests = []

        def transport(request, timeout, max_bytes):
            requests.append(request)
            payload = ({"fileUrl": "https://files.iwara.tv/file/id?expires=2000000000",
                        "file": {"height": 1080}}
                       if "/video/" in request.url else [
                           {"name": "Source", "src": {"view": "//hime.iwara.tv/1080.mp4"}},
                           {"name": "540", "src": {"view": "//mikoto.iwara.tv/540.mp4"}}])
            return HttpResponse(200, {}, json.dumps(payload).encode())

        item = SimpleNamespace(id=90, provider="iwara", external_id="Video123",
                               url="https://www.iwara.tv/video/Video123", metadata={},
                               media_url="https://www.iwara.tv/video/Video123")
        resolver = FollowMediaResolver(transport).with_credential_loader(
            lambda _: Credential("iwara", {"access_token": "private-token"}))
        self.assertEqual(resolver.resolve(item).url, "https://hime.iwara.tv/1080.mp4")
        lower = resolver.resolve(item, height=540)
        self.assertEqual(lower.url, "https://mikoto.iwara.tv/540.mp4")
        self.assertEqual(len(requests), 2)
        self.assertEqual(lower.allowed_hosts, ("iwara.tv",))
        self.assertIsNone(lower.headers)
        self.assertNotIn("Authorization", requests[1].headers)
        self.assertTrue(proxyable("iwara", item.media_url))

    def test_iwara_access_failure_is_cached_and_rate_limit_pauses_the_source(self):
        transport = mock.Mock(return_value=HttpResponse(429, {"Retry-After": "30"}, b""))
        resolver = FollowMediaResolver(transport, clock=lambda: 100)
        item = SimpleNamespace(id=90, provider="iwara", external_id="Video123",
                               url="https://www.iwara.tv/video/Video123", metadata={})
        with self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(item)
        with self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(item)
        self.assertEqual(transport.call_count, 1)
        self.assertEqual(resolver.cooling("iwara"), 30)

    def test_iwara_unknown_original_height_preserves_default_and_explicit_selection(self):
        transport = mock.Mock(side_effect=[
            HttpResponse(200, {}, json.dumps({
                "fileUrl": "https://files.iwara.tv/file/id?expires=2000000000",
                "file": {"height": 0}}).encode()),
            HttpResponse(200, {}, json.dumps([
                {"name": "540", "src": {"view": "//hime.iwara.tv/540.mp4"}},
                {"name": "Source", "src": {"view": "//mikoto.iwara.tv/original.mp4"}}]).encode())])
        resolver = FollowMediaResolver(transport)
        item = SimpleNamespace(id=91, provider="iwara", external_id="Video123",
                               url="https://www.iwara.tv/video/Video123", metadata={})
        original = resolver.resolve(item)
        self.assertEqual(original.url, "https://mikoto.iwara.tv/original.mp4")
        self.assertEqual(original.qualities[0][0], 0)
        self.assertEqual(resolver.resolve(item, height=540).url, "https://hime.iwara.tv/540.mp4")
        self.assertEqual(resolver.resolve(item, height=0), original)
        self.assertEqual(resolver.resolve(item, height=9999), original)
        self.assertEqual(transport.call_count, 2)

    def test_rule34video_resolves_the_current_signed_url_and_caches_it(self):
        requests = []

        def transport(request, timeout, max_bytes):
            requests.append(request)
            return HttpResponse(200, {}, b"<script>video_url: 'https:\\/\\/rule34video.com\\/get_file\\/1.mp4\\/?v-acctoken=fresh'</script>")

        item = SimpleNamespace(
            id=7, provider="rule34video",
            url="https://rule34video.com/video/7/work/",
            media_url="https://rule34video.com/get_file/1.mp4/",
            metadata={},
        )
        resolver = FollowMediaResolver(transport)
        first = resolver.resolve(item)
        second = resolver.resolve(item)
        self.assertEqual(first, second)
        self.assertEqual(len(requests), 1)
        self.assertIn("v-acctoken=fresh", first.url)
        self.assertEqual(first.referer, item.url)

    def test_non_rule34_media_uses_the_recorded_source_without_a_network_probe(self):
        resolver = FollowMediaResolver(lambda *_args: self.fail("unexpected network probe"))
        item = SimpleNamespace(
            id=8, provider="kemono", url="https://kemono.cr/post/8",
            media_url="https://img.kemono.cr/data/8.mp4", metadata={},
        )
        self.assertEqual(resolver.resolve(item).url, item.media_url)

    def test_simpcity_images_on_any_public_image_host_are_proxied_without_a_cookie(self):
        """帖子里的图挂在几十家图站上，白名单追不上；这类媒体不带凭据，放行任意公网主机。"""
        resolver = FollowMediaResolver(lambda *_args: self.fail("unexpected network probe"))
        item = SimpleNamespace(
            id=11, provider="simpcity", url="https://simpcity.cr/threads/17401/post-1",
            media_url="https://jpg5.su/img/abc.jpg", metadata={},
        )
        resolved = resolver.resolve(item)
        self.assertEqual(resolved.url, item.media_url)
        self.assertIsNone(resolved.headers)
        self.assertEqual(resolved.allowed_hosts, ())
        self.assertTrue(resolved.public_hosts)
        for url in ("https://simp6.cuckcapital.cr/images4/b73b.png", "https://jpg5.su/img/abc.jpg",
                    "https://i.ibb.co/x/y.png", "https://images2.imgbox.com/ab/cd/ef.jpg"):
            self.assertTrue(proxyable("simpcity", url), url)
        # 放宽的是域名清单，不是安全边界：明文、IP 字面量、本机与局域网名字、带用户
        # 信息的地址照旧拒收。
        for url in ("http://jpg5.su/img/abc.jpg", "https://169.254.1.1/a.png", "https://[::1]/a.png",
                    "https://localhost/a.png", "https://peach.local/a.png", "https://router.lan/a.png",
                    "https://intranet/a.png", "https://user:pw@jpg5.su/a.png"):
            self.assertFalse(proxyable("simpcity", url), url)
        # 别的来源仍只认自己的白名单。
        self.assertFalse(proxyable("kemono", "https://jpg5.su/img/abc.jpg"))

    def test_public_host_media_never_carries_credentials(self):
        """公网模式成立的前提是这条媒体不带凭据，结构上就不允许两者同时出现。"""
        with self.assertRaises(ValueError):
            ResolvedFollowMedia("https://jpg5.su/a.jpg", headers={"Cookie": "x"},
                                public_hosts=True)

    def test_untrusted_or_credentialed_targets_are_rejected(self):
        resolver = FollowMediaResolver(lambda *_args: self.fail("unexpected network probe"))
        bad = SimpleNamespace(
            id=9, provider="kemono", url="https://kemono.cr/post/9",
            media_url="https://127.0.0.1/private.mp4", metadata={},
        )
        with self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(bad)
        locked = SimpleNamespace(
            id=10, provider="f95zone", url="https://f95zone.to/post/10",
            media_url="https://f95zone.to/file.mp4",
            metadata={"media_needs_credential": True},
        )
        with self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(locked)

    def test_fanbox_gallery_selects_an_image_by_index(self):
        item = SimpleNamespace(
            id=11, provider="fanbox", url="https://creator.fanbox.cc/posts/11",
            media_url=None, metadata={"media_items": [
                {"url": "https://downloads.fanbox.cc/one.jpg",
                 "resource_provider": "fanbox"},
                {"url": "https://downloads.fanbox.cc/two.jpg",
                 "resource_provider": "fanbox"},
            ]},
        )
        target = FollowMediaResolver(lambda *_args: None).resolve(item, 1)
        self.assertEqual(target.url, "https://downloads.fanbox.cc/two.jpg")

    def test_archive_attachment_list_resolves_only_on_the_items_own_site(self):
        """归档站帖子的附件清单按条目自己站点的主机白名单取，别的主机、别的来源一律拒收。"""
        item = SimpleNamespace(
            id=15, provider="pawchive", url="https://pawchive.pw/patreon/user/1/post/15",
            media_url="https://file.pawchive.pw/data/a/clip.mp4", metadata={"media_items": [
                {"url": "https://file.pawchive.pw/data/a/clip.mp4", "resource_provider": "pawchive"},
                {"url": "https://file.pawchive.pw/data/a/page2.jpg", "resource_provider": "pawchive"},
                {"url": "https://127.0.0.1/data/a/private.jpg", "resource_provider": "pawchive"},
                {"url": "https://kemono.cr/data/a/page3.jpg", "resource_provider": "kemono"},
            ]},
        )
        resolver = FollowMediaResolver(lambda *_args: None)
        target = resolver.resolve(item, 1)
        self.assertEqual(target.url, "https://file.pawchive.pw/data/a/page2.jpg")
        self.assertEqual(target.referer, item.url)
        for index in (2, 3):
            with self.assertRaises(FollowMediaUnavailable):
                resolver.resolve(item, index)

    def test_legacy_f95_attachment_is_resolved_as_an_image(self):
        item = SimpleNamespace(
            id=13, provider="f95zone", url="https://f95zone.to/threads/1/post-13",
            media_url="https://pixeldrain.com/l/files", metadata={"attachments": [
                "https://attachments.f95zone.to/2026/08/one.jpg",
            ]},
        )
        target = FollowMediaResolver(lambda *_args: None).resolve(item, 0)
        self.assertEqual(target.url,
                         "https://attachments.f95zone.to/2026/08/one.jpg")
        self.assertEqual(target.referer, item.url)

    def test_f95_attachment_stays_available_while_other_links_need_a_session(self):
        item = SimpleNamespace(
            id=14, provider="f95zone", url="https://f95zone.to/threads/1/post-14",
            media_url="https://f95zone.to/masked/gofile.io/1/locked",
            metadata={"media_needs_credential": True, "attachments": [
                "https://attachments.f95zone.to/2026/08/one.jpg",
            ]},
        )
        resolver = FollowMediaResolver(lambda *_args: None).with_credential_loader(
            lambda provider: Credential(provider, {"cookie": "xf_session=saved"}))
        target = resolver.resolve(item, 0)
        self.assertEqual(target.url,
                         "https://attachments.f95zone.to/2026/08/one.jpg")
        self.assertEqual(target.headers, {"Cookie": "xf_session=saved"})

    def test_gofile_media_keeps_the_token_in_the_proxy_header(self):
        item = SimpleNamespace(
            id=12, provider="fanbox", url="https://creator.fanbox.cc/posts/12",
            media_url=None, metadata={"media_items": [
                {"url": "https://store1.gofile.io/download/one.mp4",
                 "resource_provider": "gofile"},
            ]},
        )
        resolver = FollowMediaResolver(lambda *_args: None).with_credential_loader(
            lambda provider: Credential(provider, {"api_token": "secret"}))
        target = resolver.resolve(item, 0)
        self.assertEqual(target.headers, {"Authorization": "Bearer secret"})
        self.assertNotIn("secret", target.url)

    def test_the_resolution_cache_stays_bounded(self):
        """只有 rule34video 会进这个缓存，一次列表几百条；只按 TTL 过期的话，
        进程活多久它就长多久。"""
        def transport(request, timeout, max_bytes):
            return HttpResponse(200, {}, b"<script>video_url: "
                                b"'https://rule34video.com/get_file/1_720p.mp4/'</script>")

        resolver = FollowMediaResolver(transport, max_cache_entries=2)
        for identifier in range(6):
            resolver.resolve(SimpleNamespace(
                id=identifier, provider="rule34video",
                url=f"https://rule34video.com/video/{identifier}/x/",
                media_url=None, metadata={}))
        self.assertEqual(len(resolver._cache), 2)


class Rule34VideoUpstreamDedupeTests(unittest.TestCase):
    """播放器起播时同时问正片和清晰度，之后每个 Range 再问一遍；这些都只该换来一次详情页。"""

    PAGE = b"<script>video_url: 'https://rule34video.com/get_file/1_720p.mp4/'</script>"

    @staticmethod
    def _item(identifier=7):
        return SimpleNamespace(id=identifier, provider="rule34video",
                               url=f"https://rule34video.com/video/{identifier}/x/",
                               media_url=None, metadata={})

    def test_concurrent_requests_for_one_item_share_a_single_detail_fetch(self):
        requests = []
        entered = threading.Event()
        release = threading.Event()

        def transport(request, timeout, max_bytes):
            requests.append(request.url)
            entered.set()
            release.wait(5)
            return HttpResponse(200, {}, self.PAGE)

        resolver = FollowMediaResolver(transport)
        results = []
        workers = [threading.Thread(target=lambda: results.append(resolver.resolve(self._item())))
                   for _ in range(3)]
        workers[0].start()
        self.assertTrue(entered.wait(5))
        for worker in workers[1:]:
            worker.start()
        release.set()
        for worker in workers:
            worker.join(5)
        self.assertEqual(len(requests), 1)
        self.assertEqual(len({result.url for result in results}), 1)
        self.assertEqual(len(results), 3)

    def test_a_missing_detail_page_is_not_refetched_within_the_ttl(self):
        clock = [1000.0]
        transport = mock.Mock(return_value=HttpResponse(404, {}, b""))
        resolver = FollowMediaResolver(transport, clock=lambda: clock[0])
        for _ in range(3):
            with self.assertRaises(FollowMediaUnavailable):
                resolver.resolve(self._item())
        self.assertEqual(transport.call_count, 1)
        clock[0] += resolver.MISS_TTL + 1
        transport.return_value = HttpResponse(200, {}, self.PAGE)
        self.assertIn("1_720p", resolver.resolve(self._item()).url)
        self.assertEqual(transport.call_count, 2)

    def test_a_rate_limited_source_pauses_every_item_until_retry_after(self):
        clock = [1000.0]
        transport = mock.Mock(return_value=HttpResponse(429, {"retry-after": "90"}, b""))
        resolver = FollowMediaResolver(transport, clock=lambda: clock[0])
        with self.assertLogs("peach.follow_stream", "WARNING"):
            with self.assertRaises(FollowMediaUnavailable):
                resolver.resolve(self._item(7))
        with self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(self._item(8))
        self.assertEqual(transport.call_count, 1)
        self.assertGreater(resolver.cooling("rule34video"), 0)
        clock[0] += 91
        self.assertEqual(resolver.cooling("rule34video"), 0)
        transport.return_value = HttpResponse(200, {}, self.PAGE)
        self.assertIn("1_720p", resolver.resolve(self._item(8)).url)
        self.assertEqual(transport.call_count, 2)

    def test_retry_after_is_capped_and_missing_retry_after_uses_the_default(self):
        clock = [0.0]
        resolver = FollowMediaResolver(
            mock.Mock(return_value=HttpResponse(403, {"Retry-After": "999999"}, b"")),
            clock=lambda: clock[0])
        with self.assertLogs("peach.follow_stream", "WARNING"), self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(self._item(7))
        self.assertEqual(resolver.cooling("rule34video"), resolver.MAX_COOLDOWN)
        resolver = FollowMediaResolver(mock.Mock(return_value=HttpResponse(429, {}, b"")),
                                       clock=lambda: clock[0])
        with self.assertLogs("peach.follow_stream", "WARNING"), self.assertRaises(FollowMediaUnavailable):
            resolver.resolve(self._item(7))
        self.assertEqual(resolver.cooling("rule34video"), resolver.COOLDOWN)


class Rule34VideoQualityTests(unittest.TestCase):
    """rule34video 把每档清晰度写成独立字段，只解析 video_url 会永远播最低那档。

    2026-08-31 实测 video/4564733 的页面给出四档：

        video_url      -> 4564733_360.mp4
        video_alt_url  -> 4564733_480p.mp4
        video_alt_url2 -> 4564733_720p.mp4
        video_alt_url3 -> 4564733_1080p.mp4
    """

    PAGE = (b"<script>"
            b"video_url: 'https://rule34video.com/get_file/51/a/1/1_360.mp4/?v-acctoken=t0';"
            b"video_alt_url: 'https://rule34video.com/get_file/51/a/1/1_480p.mp4/?v-acctoken=t1';"
            b"video_alt_url2: 'https://rule34video.com/get_file/51/a/1/1_720p.mp4/?v-acctoken=t2';"
            b"video_alt_url3: 'https://rule34video.com/get_file/51/a/1/1_1080p.mp4/?v-acctoken=t3';"
            b"</script>")

    def _resolver(self, body=None):
        def transport(request, timeout, max_bytes):
            return HttpResponse(200, {}, self.PAGE if body is None else body)
        return FollowMediaResolver(transport)

    def _item(self):
        return SimpleNamespace(
            id=11, provider="rule34video",
            url="https://rule34video.com/video/11/x/",
            media_url="https://rule34video.com/get_file/1.mp4/",
            metadata={},
        )

    def test_every_quality_is_parsed_and_the_highest_plays_by_default(self):
        resolved = self._resolver().resolve(self._item())
        self.assertEqual([height for height, _ in resolved.qualities],
                         [1080, 720, 480, 360])
        self.assertIn("1_1080p.mp4", resolved.url,
                      "默认应当是最高一档，旧实现固定播 360")

    def test_a_requested_height_is_honoured(self):
        resolved = self._resolver().resolve(self._item(), None, 480)
        self.assertIn("1_480p.mp4", resolved.url)

    def test_an_unavailable_height_falls_back_instead_of_failing(self):
        """签名 URL 会过期、来源也会改版；选档失败不该中断播放。"""
        resolved = self._resolver().resolve(self._item(), None, 4320)
        self.assertIn("1_1080p.mp4", resolved.url)

    def test_other_providers_report_no_quality_choices(self):
        item = SimpleNamespace(
            id=12, provider="pawchive",
            url="https://pawchive.pw/x",
            media_url="https://pawchive.pw/data/a/b/c.mp4",
            metadata={},
        )
        resolved = self._resolver(body=b"").resolve(item)
        self.assertEqual(resolved.qualities, (),
                         "只有 rule34video 给多档，其余来源只显示原画")


class ProxyUpstreamTests(unittest.TestCase):
    """`/follow-stream` 的代理边界：跟哪些跳、转发哪些头、什么时候干脆不转发。

    生产用的那个 client 开着 `follow_redirects=True`，所以这里的替身也开着：
    要证明的正是「按跳自己判」而不是「靠 client 的默认值」。
    """

    def _client(self, handler):
        client = httpx.Client(transport=httpx.MockTransport(handler),
                              follow_redirects=True)
        self.addCleanup(client.close)
        return client

    def test_a_redirect_out_of_the_whitelist_never_carries_the_credential(self):
        """上游只要回一个指向别处的 Location，httpx 就会把 Cookie 一路带过去。"""
        seen = []

        def handler(request):
            seen.append(request)
            return httpx.Response(302, request=request, headers={
                "location": "https://collector.example/steal.jpg"})

        target = ResolvedFollowMedia(
            "https://attachments.f95zone.to/2026/08/one.jpg",
            "https://f95zone.to/threads/1/post-1",
            {"Cookie": "xf_session=fixture"},
            allowed_hosts=("attachments.f95zone.to",))
        with self.assertRaises(FollowProxyError):
            open_upstream(self._client(handler), "GET", target, incoming={})
        self.assertEqual([str(request.url) for request in seen],
                         ["https://attachments.f95zone.to/2026/08/one.jpg"],
                         "出界的那一跳一次都不能发出去")
        self.assertEqual(seen[0].headers["cookie"], "xf_session=fixture",
                         "第一跳确实带着会话，所以拦住的是真的泄露")

    def test_a_redirect_inside_the_source_domain_is_still_followed(self):
        """归档站主域会 302 到取文件的节点（实测 kemono → n1.），必须跟得上。"""

        def handler(request):
            if request.url.host == "kemono.cr":
                return httpx.Response(302, request=request, headers={
                    "location": "https://n1.kemono.cr/data/7.mp4"})
            return httpx.Response(200, request=request,
                                  stream=httpx.ByteStream(b"frames"),
                                  headers={"content-type": "video/mp4"})

        target = ResolvedFollowMedia("https://kemono.cr/data/7.mp4",
                                     allowed_hosts=("kemono.cr",))
        upstream = open_upstream(self._client(handler), "GET", target, incoming={})
        self.addCleanup(upstream.close)
        self.assertEqual(upstream.read(), b"frames")

    def test_rule34video_is_followed_out_to_its_cdn(self):
        """rule34video 的 `/get_file/…` 只是签名入口，正片在 `*.boomio-cdn.com`。

        这一跳没有凭据可泄露（这条媒体不带 `headers`），拦住它的唯一后果是每条
        视频都停在 502。CDN 后缀登记在 `follow_providers` 的 `hosts` 里，代理照样
        逐跳校验。
        """

        def handler(request):
            if request.url.host == "rule34video.com":
                return httpx.Response(302, request=request, headers={
                    "location": "https://eu-cdn05-prem.boomio-cdn.com/x.mp4"})
            return httpx.Response(206, request=request,
                                  stream=httpx.ByteStream(b"frames"),
                                  headers={"content-type": "video/mp4"})

        target = ResolvedFollowMedia(
            "https://rule34video.com/get_file/1/1080p.mp4/?v-acctoken=fresh",
            "https://rule34video.com/video/1/work/",
            allowed_hosts=("rule34video.com", "boomio-cdn.com"))
        upstream = open_upstream(self._client(handler), "GET", target, incoming={})
        self.addCleanup(upstream.close)
        self.assertEqual(upstream.read(), b"frames")

    def test_an_upstream_error_page_is_refused_instead_of_forwarded(self):
        """403 页面、限流提示、错误 JSON 转发给播放器没有用处，只会把上游的
        主机名和提示语抄给浏览器。"""

        def handler(request):
            return httpx.Response(403, request=request,
                                  content=b"<html>blocked by upstream shield</html>")

        target = ResolvedFollowMedia("https://kemono.cr/data/7.mp4",
                                     allowed_hosts=("kemono.cr",))
        with self.assertRaises(FollowProxyError) as raised:
            open_upstream(self._client(handler), "GET", target, incoming={})
        self.assertIn("403", str(raised.exception))
        self.assertEqual(raised.exception.status_code, 403)
        self.assertNotIn("shield", str(raised.exception))

    def test_a_redirect_loop_is_bounded(self):
        seen = []

        def handler(request):
            seen.append(request)
            return httpx.Response(302, request=request, headers={
                "location": "https://kemono.cr/data/next.mp4"})

        target = ResolvedFollowMedia("https://kemono.cr/data/7.mp4",
                                     allowed_hosts=("kemono.cr",))
        with self.assertRaises(FollowProxyError):
            open_upstream(self._client(handler), "GET", target, incoming={})
        self.assertEqual(len(seen), MAX_PROXY_REDIRECTS + 1)

    def test_media_without_a_whitelist_is_not_proxied_at_all(self):
        target = ResolvedFollowMedia("https://kemono.cr/data/7.mp4")
        with self.assertRaises(FollowProxyError):
            open_upstream(
                self._client(lambda request: self.fail("不该发出请求")),
                "GET", target, incoming={})

    def test_public_host_media_follows_redirects_across_public_hosts(self):
        """图站常把原图 302 到自己的 CDN，公网模式下这一跳要跟得上。"""
        def handler(request):
            if request.url.host == "jpg5.su":
                return httpx.Response(302, request=request, headers={
                    "location": "https://cdn.jpg5cdn.example/a.jpg"})
            return httpx.Response(200, request=request,
                                  stream=httpx.ByteStream(b"pixels"),
                                  headers={"content-type": "image/jpeg"})

        target = ResolvedFollowMedia("https://jpg5.su/a.jpg", public_hosts=True)
        with mock.patch.object(peach_http, "host_addresses", return_value=("93.184.216.34",)):
            upstream = open_upstream(self._client(handler), "GET", target, incoming={})
        self.addCleanup(upstream.close)
        self.assertEqual(upstream.read(), b"pixels")
        self.assertNotIn("cookie", upstream.request.headers)

    def test_public_host_media_never_reaches_into_the_lan(self):
        """域名字面上像公网但解析到内网，或者上游把我们指到内网地址：一次都不发。"""
        seen = []

        def to_lan(request):
            seen.append(request)
            return httpx.Response(302, request=request, headers={
                "location": "https://169.254.169.254/admin.png"})

        target = ResolvedFollowMedia("https://jpg5.su/a.jpg", public_hosts=True)
        with mock.patch.object(peach_http, "host_addresses", return_value=("93.184.216.34",)):
            with self.assertRaises(FollowProxyError):
                open_upstream(self._client(to_lan), "GET", target, incoming={})
        self.assertEqual([str(request.url) for request in seen], ["https://jpg5.su/a.jpg"])
        for addresses in ((), ("127.0.0.1",), ("93.184.216.34", "fd00::1")):
            with mock.patch.object(peach_http, "host_addresses", return_value=addresses):
                with self.assertRaises(FollowProxyError, msg=str(addresses)):
                    open_upstream(self._client(lambda request: self.fail("不该发出请求")),
                                  "GET", target, incoming={})

    def test_only_range_headers_come_in_and_only_media_headers_go_back(self):
        target = ResolvedFollowMedia(
            "https://kemono.cr/data/7.mp4", "https://kemono.cr/post/7",
            {"Cookie": "source_session=fixture"}, allowed_hosts=("kemono.cr",))
        headers = proxy_request_headers(target, {
            "range": "bytes=0-1", "accept": "video/*",
            # 浏览器发给 Peach 的凭据不能跟着请求送到上游去。
            "cookie": "peach=token", "authorization": "Bearer peach",
        })
        self.assertEqual(headers["Range"], "bytes=0-1")
        self.assertEqual(headers["Accept"], "video/*")
        self.assertEqual(headers["Referer"], "https://kemono.cr/post/7")
        self.assertEqual(headers["Cookie"], "source_session=fixture")
        self.assertNotIn("Authorization", headers)