diff --git a/tests/unit/test_pagination.py b/tests/unit/test_pagination.py index 9d9dcc7c1..1d7c55e30 100755 --- a/tests/unit/test_pagination.py +++ b/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 diff --git a/tests/unit/test_text_splitter.py b/tests/unit/test_text_splitter.py index d52bec9a0..c87a3ad72 100755 --- a/tests/unit/test_text_splitter.py +++ b/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 diff --git a/tests/unit/test_url_security.py b/tests/unit/test_url_security.py index 20bdcb026..c10eac965 100755 --- a/tests/unit/test_url_security.py +++ b/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)