tests/unit/test_pagination.py

@@ -242,3 +242,147 @@ class TestPaginateFunction:
 
         assert len(result.data) == 2
         assert result.data[0]["id"] == 1
+
+
+# ── PaginationParams 补充边界 ───────────────────────────────────────────────
+
+
+class TestPaginationParamsEdgeCases:
+    """PaginationParams 补充边界场景."""
+
+    def test_page_size_1_minimum(self):
+        """page_size=1 是允许的最小值."""
+        params = PaginationParams(page_size=1)
+        assert params.page_size == 1
+        assert params.limit == 1
+
+    def test_page_size_100_maximum(self):
+        """page_size=100 是允许的最大值."""
+        params = PaginationParams(page_size=100)
+        assert params.page_size == 100
+
+    def test_offset_page_1_size_100(self):
+        """第1页每页100条 offset=0."""
+        params = PaginationParams(page=1, page_size=100)
+        assert params.offset == 0
+
+    def test_offset_page_100_size_100(self):
+        """第100页每页100条 offset=9900."""
+        params = PaginationParams(page=100, page_size=100)
+        assert params.offset == 9900
+
+    def test_large_page_number_accepted(self):
+        """极大页码（超过实际页数）允许."""
+        params = PaginationParams(page=999999, page_size=20)
+        assert params.page == 999999
+        assert params.offset == (999999 - 1) * 20
+
+
+# ── PaginationMeta 补充边界 ─────────────────────────────────────────────────
+
+
+class TestPaginationMetaEdgeCases:
+    """PaginationMeta 补充边界场景."""
+
+    def test_total_0_page_1(self):
+        """total=0, page=1 时 total_pages=0, 无上下页."""
+        params = PaginationParams(page=1, page_size=20)
+        meta = PaginationMeta.from_params(params, total=0)
+        assert meta.total_pages == 0
+        assert meta.has_next is False
+        assert meta.has_prev is False
+
+    def test_total_0_page_beyond(self):
+        """total=0, page>1 时 has_prev=True（因为page>1）."""
+        params = PaginationParams(page=3, page_size=20)
+        meta = PaginationMeta.from_params(params, total=0)
+        assert meta.total_pages == 0
+        assert meta.has_next is False
+        assert meta.has_prev is True
+
+    def test_exact_last_page(self):
+        """刚好是最后一页时 has_next=False."""
+        params = PaginationParams(page=5, page_size=10)
+        meta = PaginationMeta.from_params(params, total=50)
+        assert meta.total_pages == 5
+        assert meta.has_next is False
+        assert meta.has_prev is True
+
+    def test_one_more_than_exact(self):
+        """比整数页多1条时总页数+1."""
+        params = PaginationParams(page=1, page_size=10)
+        meta = PaginationMeta.from_params(params, total=51)
+        assert meta.total_pages == 6
+
+    def test_page_exactly_total_pages(self):
+        """page == total_pages 时 has_next=False."""
+        params = PaginationParams(page=3, page_size=10)
+        meta = PaginationMeta.from_params(params, total=30)
+        assert meta.has_next is False
+
+    def test_total_1_page_1_size_1(self):
+        """1条数据1页."""
+        params = PaginationParams(page=1, page_size=1)
+        meta = PaginationMeta.from_params(params, total=1)
+        assert meta.total_pages == 1
+        assert meta.has_next is False
+        assert meta.has_prev is False
+
+
+# ── paginate 补充边界 ──────────────────────────────────────────────────────
+
+
+class TestPaginateEdgeCases:
+    """paginate 补充边界场景."""
+
+    def test_single_item_list(self):
+        """单元素列表."""
+        result = paginate([42], PaginationParams(page=1, page_size=10))
+        assert result.data == [42]
+        assert result.pagination.total == 1
+        assert result.pagination.total_pages == 1
+
+    def test_page_exactly_last(self):
+        """刚好在最后一页."""
+        items = list(range(25))
+        result = paginate(items, PaginationParams(page=3, page_size=10))
+        assert result.data == list(range(20, 25))
+        assert result.pagination.has_next is False
+
+    def test_page_past_end_returns_empty(self):
+        """页码超过总数返回空."""
+        items = list(range(5))
+        result = paginate(items, PaginationParams(page=10, page_size=10))
+        assert result.data == []
+        assert result.pagination.total == 5
+
+    def test_empty_list_page_1(self):
+        """空列表第1页."""
+        result = paginate([], PaginationParams(page=1, page_size=10))
+        assert result.data == []
+        assert result.pagination.total == 0
+        assert result.pagination.total_pages == 0
+
+    def test_page_size_1_iterates_all(self):
+        """page_size=1 时每页1条."""
+        items = ["a", "b", "c"]
+        r1 = paginate(items, PaginationParams(page=1, page_size=1))
+        r2 = paginate(items, PaginationParams(page=2, page_size=1))
+        r3 = paginate(items, PaginationParams(page=3, page_size=1))
+        assert r1.data == ["a"]
+        assert r2.data == ["b"]
+        assert r3.data == ["c"]
+
+    def test_does_not_mutate_input(self):
+        """不修改输入列表."""
+        items = [1, 2, 3, 4, 5]
+        original = items[:]
+        paginate(items, PaginationParams(page=1, page_size=2))
+        assert items == original
+
+    def test_page_size_greater_than_total(self):
+        """每页条数大于总数."""
+        items = list(range(5))
+        result = paginate(items, PaginationParams(page=1, page_size=100))
+        assert result.data == items
+        assert result.pagination.total_pages == 1

tests/unit/test_text_splitter.py

@@ -145,3 +145,183 @@ class TestSplitText:
 
         for seg in result:
             assert len(seg) <= 80
+
+
+# ── 短文本与空文本补充 ──────────────────────────────────────────────────────
+
+
+class TestSplitTextEmptyAndShort:
+    """空文本与短文本补充场景."""
+
+    def test_whitespace_only_returns_empty(self):
+        """纯空白文本返回空列表."""
+        assert split_text("   \n\t  ") == []
+
+    def test_single_char(self):
+        """单字符文本."""
+        assert split_text("好", max_chars=10) == ["好"]
+
+    def test_exactly_max_chars_no_split(self):
+        """刚好等于 max_chars 不分割."""
+        text = "a" * 100
+        result = split_text(text, max_chars=100)
+        assert len(result) == 1
+        assert result[0] == text
+
+    def test_one_over_max_chars_splits(self):
+        """超过 max_chars 1 个字符就会分割."""
+        text = "a" * 101
+        result = split_text(text, max_chars=100)
+        assert len(result) >= 2
+
+    def test_none_raises(self):
+        """None 输入抛 AttributeError（strip 失败）."""
+        with pytest.raises(AttributeError):
+            split_text(None)
+
+
+# ── 句子边界分段补充 ──────────────────────────────────────────────────────
+
+
+class TestSplitTextSentenceBoundaries:
+    """句子边界分段补充场景."""
+
+    def test_split_on_fullwidth_period(self):
+        """全角句号分段."""
+        text = "第一句很长的内容。" * 20
+        result = split_text(text, max_chars=60)
+        assert len(result) > 1
+        for seg in result:
+            assert len(seg) <= 60
+
+    def test_split_on_fullwidth_question(self):
+        """全角问号分段."""
+        text = "你知道这是为什么吗？" + "是的。" * 20
+        result = split_text(text, max_chars=60)
+        assert len(result) > 1
+
+    def test_split_on_fullwidth_exclamation(self):
+        """全角感叹号分段."""
+        text = "真是太棒了！" + "内容。" * 20
+        result = split_text(text, max_chars=60)
+        assert len(result) > 1
+
+    def test_split_on_newline(self):
+        """换行符分段."""
+        lines = ["这是第一行很长的一段文字内容" * 3 for _ in range(5)]
+        text = "\n".join(lines)
+        result = split_text(text, max_chars=80)
+        assert len(result) > 1
+
+    def test_split_on_semicolon(self):
+        """全角分号分段."""
+        text = "第一项内容；" + "其他内容。" * 20
+        result = split_text(text, max_chars=60)
+        assert len(result) > 1
+
+    def test_english_period_splits(self):
+        """英文句号分段."""
+        text = "Hello world. " * 30
+        result = split_text(text, max_chars=80)
+        assert len(result) > 1
+
+    def test_short_sentences_stay_merged(self):
+        """短句（都 < 50字的句子不会单独成段，会累积到一起."""
+        text = "你好。我好。大家好。"
+        result = split_text(text, max_chars=200)
+        assert len(result) == 1
+
+
+# ── 长句强制切段补充 ──────────────────────────────────────────────────────
+
+
+class TestSplitTextLongSentenceForce:
+    """超长单句强制切段补充."""
+
+    def test_no_punctuation_forced_split(self):
+        """完全没有标点的超长文本硬切."""
+        text = "字" * 300
+        result = split_text(text, max_chars=100)
+        assert len(result) == 3
+        for seg in result:
+            assert len(seg) == 100
+
+    def test_force_split_preserves_content(self):
+        """硬切不丢字符."""
+        text = "a" * 250
+        result = split_text(text, max_chars=100)
+        assert sum(len(s) for s in result) == 250
+
+    def test_mixed_long_and_short(self):
+        """长句短句混合."""
+        long_part = "非常长的句子没有标点符号" * 15
+        text = long_part + "。结尾。"
+        result = split_text(text, max_chars=100)
+        assert len(result) > 1
+        for seg in result:
+            assert len(seg) <= 100
+
+
+# ── 短段合并补充 ─────────────────────────────────────────────────────────
+
+
+class TestSplitTextShortSegmentMerge:
+    """短段合并补充场景."""
+
+    def test_multiple_short_sentences_merged(self):
+        """多个短句合并成一段."""
+        sentences = ["你好。", "我好。", "大家好。", "天气好。", "心情好。"]
+        text = "".join(sentences)
+        result = split_text(text, max_chars=200)
+        assert len(result) == 1
+
+    def test_short_tail_merged(self):
+        """尾部短段被合并到前一段."""
+        # 前面一段接近 max_chars，尾部很短
+        long_part = "一二三四五六七八九十" * 9 + "。"  # ~90字
+        tail = "完。"  # 2字
+        text = long_part + tail
+        result = split_text(text, max_chars=100)
+        # 尾部短的应该被合并
+        assert len(result) <= 2
+
+
+# ── 边界情况补充 ─────────────────────────────────────────────────────────
+
+
+class TestSplitTextEdgeCases:
+    """边界情况补充."""
+
+    def test_only_punctuation(self):
+        """纯标点符号."""
+        text = "。。。。。"
+        result = split_text(text, max_chars=10)
+        assert len(result) == 1
+
+    def test_mixed_chinese_english(self):
+        """中英文混合."""
+        text = "你好Hello。World!" * 20
+        result = split_text(text, max_chars=100)
+        assert len(result) > 1
+        for seg in result:
+            assert len(seg) <= 100
+
+    def test_strip_whitespace(self):
+        """首尾空白被去除."""
+        text = "   你好世界。   "
+        result = split_text(text, max_chars=100)
+        assert result == ["你好世界。"]
+
+    def test_total_length_preserved(self):
+        """分段后总长度等于原文 strip 后长度."""
+        text = "这是一段用于测试的文本内容。" * 20
+        result = split_text(text, max_chars=100)
+        assert "".join(result) == text.strip()
+
+    def test_custom_small_max_chars(self):
+        """很小的 max_chars."""
+        text = "一二三四五六七八九十。" * 5
+        result = split_text(text, max_chars=20)
+        assert len(result) > 1
+        for seg in result:
+            assert len(seg) <= 20

tests/unit/test_url_security.py

@@ -9,6 +9,7 @@ url_security URL安全校验单元测试
 - safe_download_file / safe_download_bytes: mock 网络测试
 """
 
+import ipaddress
 import os
 import tempfile
 from unittest.mock import MagicMock, patch
@@ -24,6 +25,7 @@ from packages.shared.url_security import (
     NoRedirectHandler,
     UrlSecurityError,
     _check_internal_hostnames,
+    _check_ssrf_ip,
     _is_trusted_domain,
     _validate_magic_number,
     is_url_safe,
@@ -595,3 +597,220 @@ class TestConstants:
 
     def test_max_url_length(self):
         assert MAX_URL_LENGTH == 2048
+
+
+# ── SSRF IP 检查详细覆盖 ──────────────────────────────────────────────────────
+
+
+class TestSSRFIPCheck:
+    """_check_ssrf_ip 各类型 IP 拦截覆盖."""
+
+    def test_loopback_ipv4_blocked(self):
+        """IPv4 回环 127.0.0.1 被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("127.0.0.1"))
+
+    def test_loopback_ipv4_another_blocked(self):
+        """127.x 其他段也被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("127.255.255.1"))
+
+    def test_loopback_ipv6_blocked(self):
+        """IPv6 回环 ::1 被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("::1"))
+
+    def test_private_10_range_blocked(self):
+        """10.0.0.0/8 私有段被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("10.255.255.255"))
+
+    def test_private_172_range_blocked(self):
+        """172.16.0.0/12 私有段被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("172.31.255.255"))
+
+    def test_private_192_range_blocked(self):
+        """192.168.0.0/16 私有段被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("192.168.255.255"))
+
+    def test_link_local_ipv4_blocked(self):
+        """169.254.x.x 链路本地被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("169.254.1.1"))
+
+    def test_multicast_ipv4_blocked(self):
+        """224.x 组播被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("224.0.0.251"))
+
+    def test_unspecified_ipv4_blocked(self):
+        """0.0.0.0 未指定地址被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("0.0.0.0"))
+
+    def test_unspecified_ipv6_blocked(self):
+        """:: 未指定地址被拦."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("::"))
+
+    def test_reserved_ipv4_blocked(self):
+        """240.0.0.0/4 保留段被拦（含在 is_private 或 is_reserved 中）."""
+        with pytest.raises(UrlSecurityError):
+            _check_ssrf_ip(ipaddress.ip_address("240.0.0.1"))
+
+    def test_public_ipv4_passes(self):
+        """公网 IPv4 通过 _check_ssrf_ip."""
+        _check_ssrf_ip(ipaddress.ip_address("8.8.8.8"))
+
+    def test_public_ipv4_another_passes(self):
+        """另一个公网 IPv4 通过."""
+        _check_ssrf_ip(ipaddress.ip_address("1.1.1.1"))
+
+    def test_public_ipv6_passes(self):
+        """公网 IPv6 通过."""
+        _check_ssrf_ip(ipaddress.ip_address("2001:4860:4860::8888"))
+
+
+# ── 直接 IP 访问拦截 ────────────────────────────────────────────────────────
+
+
+class TestDirectIPAccess:
+    """直接 IP 访问控制（ALLOW_DIRECT_IP 开关）."""
+
+    def test_direct_ipv4_blocked_by_default(self):
+        """默认禁止直接 IP 访问."""
+        with pytest.raises(UrlSecurityError, match="禁止直接 IP 访问"):
+            validate_url_safety("https://8.8.8.8/audio.mp3")
+
+    def test_direct_private_ip_blocked_even_with_flag(self):
+        """ALLOW_DIRECT_IP=true 时私有 IP 仍被 SSRF 拦."""
+        with patch("packages.shared.url_security.ALLOW_DIRECT_IP", True):
+            with pytest.raises(UrlSecurityError):
+                validate_url_safety("https://192.168.1.1/a.mp3")
+
+    def test_direct_ip_allowed_when_flag_on(self):
+        """ALLOW_DIRECT_IP=true 时公网 IP 通过."""
+        with patch("packages.shared.url_security.ALLOW_DIRECT_IP", True):
+            # 用 mock 绕过 DNS 解析路径，走 IP 分支
+            with patch("packages.shared.url_security._check_ssrf_ip") as mock_check:
+                result = validate_url_safety("https://8.8.8.8/a.mp3")
+                assert "8.8.8.8" in result
+                mock_check.assert_called_once()
+
+
+# ── 魔数校验补充覆盖 ────────────────────────────────────────────────────────
+
+
+class TestMagicNumberExtended:
+    """魔数校验补充：更多格式 + 边界场景."""
+
+    def test_aac_adts_mpeg4(self):
+        """AAC ADTS MPEG-4 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"\xff\xf1" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"audio/aac"})
+        finally:
+            os.unlink(tmp)
+
+    def test_m4a_ftyp_magic(self):
+        """M4A ftyp 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"\x00\x00\x00\x20ftypM4A " + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"audio/x-m4a"})
+        finally:
+            os.unlink(tmp)
+
+    def test_webp_riff_webp(self):
+        """WebP RIFF+WEBP 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"image/webp"})
+        finally:
+            os.unlink(tmp)
+
+    def test_video_mp4_magic(self):
+        """video/mp4 ftyp 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"\x00\x00\x00\x20ftypmp42" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"video/mp4"})
+        finally:
+            os.unlink(tmp)
+
+    def test_matroska_webm_magic(self):
+        """WebM EBML 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"\x1a\x45\xdf\xa3" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"video/webm"})
+        finally:
+            os.unlink(tmp)
+
+    def test_bmp_magic(self):
+        """BMP BM 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"BM" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"image/bmp"})
+        finally:
+            os.unlink(tmp)
+
+    def test_flac_magic(self):
+        """FLAC fLaC 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"fLaC" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"audio/flac"})
+        finally:
+            os.unlink(tmp)
+
+    def test_ogg_magic(self):
+        """OGG OggS 魔数通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"OggS" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"audio/ogg"})
+        finally:
+            os.unlink(tmp)
+
+    def test_read_error_raises_security_error(self):
+        """文件读取失败包装为 UrlSecurityError."""
+        with pytest.raises(UrlSecurityError, match="读取文件头失败"):
+            _validate_magic_number("/nonexistent/path/file.mp3", {"audio/mpeg"})
+
+    def test_multi_type_one_match(self):
+        """多类型白名单，只要一个匹配就通过."""
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"\xff\xd8\xff" + b"\x00" * 50)
+            tmp = f.name
+        try:
+            _validate_magic_number(tmp, {"image/png", "image/jpeg", "image/gif"})
+        finally:
+            os.unlink(tmp)
+
+    def test_application_octet_stream_skipped(self):
+        """application/octet-stream 没有专属魔数，跳过校验."""
+        # 注意：octet-stream 在 _MAGIC_NUMBERS 中没有条目，所以跳过
+        # 但实际白名单中常包含它，所以它的存在不应阻断
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(b"random stuff")
+            tmp = f.name
+        try:
+            # octet-stream + png，png有魔数，png不匹配就会失败
+            # 只有 octet-stream 时应该跳过
+            _validate_magic_number(tmp, {"application/octet-stream"})
+        finally:
+            os.unlink(tmp)