test(wave204): text_splitter + pagination 单测补全 +37测 (#1171)
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled

This commit was merged in pull request #1171.
This commit is contained in:
2026-07-30 00:26:00 +08:00
parent 77a49e3365
commit de201436ea
4 changed files with 988 additions and 1554 deletions
+292 -466
View File
@@ -1,9 +1,6 @@
"""BGM 混音纯逻辑单元测试."""
"""bgm_mixer_pure 单元测试."""
from __future__ import annotations
import pytest
from video_processing.bgm_mixer_pure import (
from apps.worker.video_processing.bgm_mixer_pure import (
BGMPureConfig,
build_bgm_filter_chain,
build_sidechain_mix_filter,
@@ -17,408 +14,286 @@ from video_processing.bgm_mixer_pure import (
validate_bgm_config,
)
# ─────────────────────────────────────────────────────────────────────────────
# should_loop_bgm 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── BGMPureConfig ────────────────────────────────────────────────────────────
class TestShouldLoopBGM:
"""BGM 循环判断测试."""
class TestBGMPureConfig:
def test_default_values(self):
cfg = BGMPureConfig()
assert cfg.volume == 0.3
assert cfg.fade_in == 0.0
assert cfg.fade_out == 0.0
assert cfg.loop_enabled is True
assert cfg.sidechain_enabled is False
assert cfg.sidechain_ratio == 0.3
assert cfg.sidechain_attack == 0.02
assert cfg.sidechain_release == 0.5
assert cfg.sidechain_threshold == -25.0
def test_need_loop_when_much_shorter(self):
"""BGM 远短于目标时长,需要循环."""
assert should_loop_bgm(10, 100, True) is True
def test_custom_values(self):
cfg = BGMPureConfig(
volume=0.5,
fade_in=1.0,
fade_out=2.0,
loop_enabled=False,
sidechain_enabled=True,
sidechain_ratio=0.5,
)
assert cfg.volume == 0.5
assert cfg.loop_enabled is False
assert cfg.sidechain_enabled is True
assert cfg.sidechain_ratio == 0.5
def test_no_loop_when_long_enough(self):
"""BGM 够长,不需要循环."""
assert should_loop_bgm(100, 100, True) is False
def test_no_loop_when_just_slightly_shorter(self):
"""BGM 只差一点点(>90%),不循环."""
assert should_loop_bgm(95, 100, True) is False
# ── should_loop_bgm ─────────────────────────────────────────────────────────
def test_threshold_90_percent(self):
"""刚好 90% 阈值,不循环(<90% 才循环)."""
assert should_loop_bgm(90, 100, True) is False
def test_just_below_threshold(self):
"""略低于 90%,需要循环."""
assert should_loop_bgm(89, 100, True) is True
class TestShouldLoopBgm:
def test_loop_enabled_much_shorter(self):
# BGM 10秒,目标60秒 → 需要循环
assert should_loop_bgm(10, 60) is True
def test_loop_disabled(self):
"""禁用循环,即使 BGM 很短也不循环."""
assert should_loop_bgm(10, 100, False) is False
assert should_loop_bgm(10, 60, loop_enabled=False) is False
def test_bgm_longer_than_target(self):
# BGM 100秒,目标60秒 → 不需要循环
assert should_loop_bgm(100, 60) is False
def test_bgm_slightly_shorter_no_loop(self):
# BGM 58秒,目标60秒 → 58 > 60*0.9=54,不需要循环
assert should_loop_bgm(58, 60) is False
def test_bgm_significantly_shorter_loops(self):
# BGM 50秒,目标60秒 → 50 < 54,需要循环
assert should_loop_bgm(50, 60) is True
def test_zero_bgm_duration(self):
"""BGM 时长为 0,不循环."""
assert should_loop_bgm(0, 100, True) is False
assert should_loop_bgm(0, 60) is False
def test_negative_bgm_duration(self):
"""BGM 时长为负,不循环."""
assert should_loop_bgm(-5, 100, True) is False
assert should_loop_bgm(-1, 60) is False
def test_zero_target_duration(self):
"""目标时长为 0,不循环."""
assert should_loop_bgm(10, 0, True) is False
assert should_loop_bgm(10, 0) is False
def test_negative_target_duration(self):
"""目标时长为负,不循环."""
assert should_loop_bgm(10, -10, True) is False
assert should_loop_bgm(10, -1) is False
def test_exact_90_percent_no_loop(self):
# 边界:bgm == target * 0.9 → 不小于,不循环
assert should_loop_bgm(54, 60) is False
# ─────────────────────────────────────────────────────────────────────────────
# calculate_loop_count 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── calculate_loop_count ────────────────────────────────────────────────────
class TestCalculateLoopCount:
"""循环次数计算测试."""
def test_exact_fit_returns_1(self):
assert calculate_loop_count(60, 60) == 1
def test_exact_multiple(self):
"""刚好整数倍."""
# 100/10 = 10, +2 = 12
assert calculate_loop_count(10, 100) == 12
def test_bgm_longer_returns_1(self):
assert calculate_loop_count(100, 60) == 1
def test_not_exact_multiple(self):
"""不是整数倍."""
# 100/30 = 3, +2 = 5
assert calculate_loop_count(30, 100) == 5
def test_needs_3_loops_plus_2_margin(self):
# 60/20 = 3 + 2 = 5
assert calculate_loop_count(20, 60) == 5
def test_bgm_longer_than_target(self):
"""BGM 比目标长,至少 1 次."""
assert calculate_loop_count(200, 100) == 1
def test_needs_2_loops_plus_2_margin(self):
# 60/30 = 2 + 2 = 4
assert calculate_loop_count(30, 60) == 4
def test_zero_bgm_duration(self):
"""BGM 时长为 0,返回 1."""
assert calculate_loop_count(0, 100) == 1
assert calculate_loop_count(0, 60) == 1
def test_negative_bgm_duration(self):
"""BGM 时长为负,返回 1."""
assert calculate_loop_count(-5, 100) == 1
assert calculate_loop_count(-1, 60) == 1
def test_zero_target_duration(self):
"""目标时长为 0,返回 1."""
assert calculate_loop_count(10, 0) == 1
def test_negative_target_duration(self):
"""目标时长为负,返回 1."""
assert calculate_loop_count(10, -10) == 1
assert calculate_loop_count(10, -1) == 1
def test_very_short_bgm(self):
"""非常短的 BGM,循环次数多."""
# 100/1 = 100, +2 = 102
assert calculate_loop_count(1, 100) == 102
def test_minimum_is_1(self):
assert calculate_loop_count(10, 5) == 1
# ─────────────────────────────────────────────────────────────────────────────
# build_bgm_filter_chain 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── build_bgm_filter_chain ──────────────────────────────────────────────────
class TestBuildBGMFilterChain:
"""BGM 预处理滤镜链构建测试."""
class TestBuildBgmFilterChain:
def test_basic_structure(self):
result = build_bgm_filter_chain(100, 60)
parts = result.split(",")
# 至少有 atrim + asetpts
assert any("atrim=" in p for p in parts)
assert "asetpts=N/SR/TB" in parts
def test_basic_volume_only(self):
"""只有音量调节."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=0.5,
)
def test_volume_filter_applied(self):
result = build_bgm_filter_chain(100, 60, volume=0.5)
assert "volume=0.500" in result
assert "aloop" not in result
assert "afade=t=in" not in result
assert "afade=t=out" not in result
assert "atrim=0:100.000" in result
assert "asetpts=N/SR/TB" in result
def test_with_loop(self):
"""需要循环的情况."""
result = build_bgm_filter_chain(
bgm_duration=10,
target_duration=100,
volume=0.3,
loop_enabled=True,
)
assert "aloop=loop=" in result
assert "volume=0.300" in result
def test_volume_one_omitted(self):
result = build_bgm_filter_chain(100, 60, volume=1.0)
assert "volume=" not in result
def test_no_loop_when_disabled(self):
"""禁用循环,即使 BGM 短也不循环."""
result = build_bgm_filter_chain(
bgm_duration=10,
target_duration=100,
volume=0.3,
loop_enabled=False,
)
assert "aloop" not in result
def test_volume_clamped(self):
# volume=2.0钳制到1.0,1.0等于默认值所以被跳过
result = build_bgm_filter_chain(100, 60, volume=2.0)
assert "volume=" not in result # 钳制到1.0后与默认相同,跳过
# 用0.5验证音量过滤器本身存在
result2 = build_bgm_filter_chain(100, 60, volume=0.5)
assert "volume=0.500" in result2
def test_fade_in_only(self):
"""只有淡入."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=1.0,
fade_in=2.5,
)
assert "afade=t=in:st=0:d=2.500" in result
assert "afade=t=out" not in result
assert "volume=" not in result # volume=1.0 不加
def test_volume_zero(self):
result = build_bgm_filter_chain(100, 60, volume=0.0)
assert "volume=0.000" in result
def test_fade_out_only(self):
"""只有淡出."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=1.0,
fade_out=3.0,
)
assert "afade=t=out:st=97.000:d=3.000" in result
assert "afade=t=in" not in result
def test_fade_in_and_out(self):
"""淡入+淡出."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=1.0,
fade_in=1.5,
fade_out=2.0,
)
def test_fade_in_applied(self):
result = build_bgm_filter_chain(100, 60, fade_in=1.5)
assert "afade=t=in:st=0:d=1.500" in result
assert "afade=t=out:st=98.000:d=2.000" in result
def test_volume_1_0_skipped(self):
"""音量为 1.0 时不添加 volume 滤镜."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=1.0,
)
assert "volume=" not in result
def test_fade_in_zero_skipped(self):
result = build_bgm_filter_chain(100, 60, fade_in=0)
assert "afade=t=in" not in result
def test_volume_0(self):
"""音量为 0."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=0.0,
)
assert "volume=0.000" in result
def test_fade_out_applied(self):
result = build_bgm_filter_chain(100, 60, fade_out=2.0)
assert "afade=t=out:st=58.000:d=2.000" in result
def test_volume_clamped_high(self):
"""音量超过 1.0 被钳制."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=1.5,
)
assert "volume=1.000" not in result # 1.0不加
# 钳制到1.0后和1.0一样,不加volume滤镜
# 但因为abs(1.0 - 1.0) < 0.001,所以不添加
assert "volume=" not in result
def test_volume_clamped_low(self):
"""音量为负被钳制到 0."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=100,
volume=-0.5,
)
assert "volume=0.000" in result
def test_fade_out_longer_than_duration(self):
"""淡出时长超过总时长,不加淡出."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=10,
volume=1.0,
fade_out=20.0,
)
def test_fade_out_longer_than_target_skipped(self):
result = build_bgm_filter_chain(100, 10, fade_out=20)
# fade_out >= safe_target,不做淡出
assert "afade=t=out" not in result
def test_fade_out_equal_to_duration(self):
"""淡出时长等于总时长,不加淡出."""
result = build_bgm_filter_chain(
bgm_duration=200,
target_duration=10,
volume=1.0,
fade_out=10.0,
)
assert "afade=t=out" not in result
def test_loop_applied_when_needed(self):
result = build_bgm_filter_chain(10, 60)
assert "aloop=loop=" in result
def test_zero_target_duration_fallback(self):
"""目标时长为 0,兜底 5 秒."""
result = build_bgm_filter_chain(
bgm_duration=3,
target_duration=0,
volume=0.5,
)
def test_no_loop_when_bgm_long(self):
result = build_bgm_filter_chain(100, 60)
assert "aloop=" not in result
def test_loop_disabled(self):
result = build_bgm_filter_chain(10, 60, loop_enabled=False)
assert "aloop=" not in result
def test_trim_to_target_duration(self):
result = build_bgm_filter_chain(100, 60)
assert "atrim=0:60.000" in result
def test_zero_target_uses_fallback(self):
result = build_bgm_filter_chain(100, 0)
# 兜底5秒
assert "atrim=0:5.000" in result
def test_negative_target_duration_fallback(self):
"""目标时长为负,兜底 5 秒."""
result = build_bgm_filter_chain(
bgm_duration=3,
target_duration=-5,
volume=0.5,
)
def test_negative_target_uses_fallback(self):
result = build_bgm_filter_chain(100, -5)
assert "atrim=0:5.000" in result
def test_full_chain_with_all_effects(self):
"""完整滤镜链:循环+音量+淡入淡出+截断+重置."""
def test_all_features_combined(self):
result = build_bgm_filter_chain(
bgm_duration=10,
target_duration=100,
volume=0.4,
bgm_duration=15,
target_duration=60,
volume=0.3,
fade_in=1.0,
fade_out=2.0,
loop_enabled=True,
)
parts = result.split(",")
# 顺序:aloop -> volume -> afade in -> afade out -> atrim -> asetpts
assert len(parts) >= 6
assert "aloop" in parts[0]
assert "volume" in parts[1]
assert "afade=t=in" in parts[2]
assert "afade=t=out" in parts[3]
assert "atrim" in parts[4]
assert "asetpts" in parts[5]
assert "aloop=loop=" in result
assert "volume=0.300" in result
assert "afade=t=in" in result
assert "afade=t=out" in result
assert "atrim=0:60.000" in result
assert "asetpts=N/SR/TB" in result
# ─────────────────────────────────────────────────────────────────────────────
# calculate_sidechain_ratio 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── calculate_sidechain_ratio ───────────────────────────────────────────────
class TestCalculateSidechainRatio:
"""Sidechain 压缩比计算测试."""
def test_zero_ratio_minimum(self):
assert calculate_sidechain_ratio(0) == 2.0
def test_default_ratio_0_3(self):
"""默认 0.3."""
# 1 / (1 - 0.3) = 1.428... 但下限是 2.0
assert calculate_sidechain_ratio(0.3) == pytest.approx(2.0, rel=0.01)
def test_ratio_0_5(self):
"""比例 0.5."""
# 1 / (1 - 0.5) = 2.0
assert calculate_sidechain_ratio(0.5) == pytest.approx(2.0, rel=0.01)
def test_ratio_0_8(self):
"""比例 0.8."""
# 1 / (1 - 0.8) = 5.0
assert calculate_sidechain_ratio(0.8) == pytest.approx(5.0, rel=0.01)
def test_ratio_0_9(self):
"""比例 0.9."""
# 1 / (1 - 0.9) = 10.0
assert calculate_sidechain_ratio(0.9) == pytest.approx(10.0, rel=0.01)
def test_ratio_0(self):
"""比例 0,返回下限 2.0."""
assert calculate_sidechain_ratio(0.0) == 2.0
def test_ratio_negative(self):
"""比例为负,返回下限 2.0."""
def test_negative_clamped(self):
assert calculate_sidechain_ratio(-0.5) == 2.0
def test_ratio_1_0(self):
"""比例 1.0,返回上限 10.0."""
def test_one_ratio_maximum(self):
assert calculate_sidechain_ratio(1.0) == 10.0
def test_ratio_greater_than_1(self):
"""比例超过 1.0,返回上限 10.0."""
assert calculate_sidechain_ratio(2.0) == 10.0
def test_above_one_clamped(self):
assert calculate_sidechain_ratio(1.5) == 10.0
def test_mid_value(self):
# ratio = 1/(1-0.5) = 2.0
result = calculate_sidechain_ratio(0.5)
assert abs(result - 2.0) < 0.01
def test_high_value(self):
# 1/(1-0.9) = 10 → 钳制到10
assert calculate_sidechain_ratio(0.9) == 10.0
def test_03_default(self):
# 1/(1-0.3) = 1.428... → 钳制到2.0
result = calculate_sidechain_ratio(0.3)
assert result >= 2.0
# ─────────────────────────────────────────────────────────────────────────────
# build_simple_mix_filter 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── build_simple_mix_filter ─────────────────────────────────────────────────
class TestBuildSimpleMixFilter:
"""普通混音滤镜构建测试."""
def test_contains_amix(self):
"""包含 amix."""
def test_contains_inputs_and_output(self):
result = build_simple_mix_filter()
assert "[0:a][1:a]" in result
assert "amix=inputs=2" in result
def test_contains_volume_compensation(self):
"""包含 volume=2 补偿."""
result = build_simple_mix_filter()
assert "duration=first" in result
assert "[final]" in result
assert "volume=2" in result
def test_output_label(self):
"""输出标签为 [final]."""
result = build_simple_mix_filter()
assert "[final]" in result
def test_duration_first(self):
"""duration=first,以主音频时长为准."""
result = build_simple_mix_filter()
assert "duration=first" in result
# ─────────────────────────────────────────────────────────────────────────────
# build_sidechain_mix_filter 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── build_sidechain_mix_filter ──────────────────────────────────────────────
class TestBuildSidechainMixFilter:
"""Sidechain 混音滤镜构建测试."""
def test_contains_sidechaincompress(self):
"""包含 sidechaincompress."""
result = build_sidechain_mix_filter()
assert "sidechaincompress=" in result
assert "[1:a][0:a]sidechaincompress" in result
def test_threshold_param(self):
"""threshold 参数正确."""
def test_threshold_in_db(self):
result = build_sidechain_mix_filter(threshold=-30.0)
assert "threshold=-30.0dB" in result
def test_attack_param(self):
"""attack 参数正确."""
result = build_sidechain_mix_filter(attack=0.05)
assert "attack=0.050" in result
def test_release_param(self):
"""release 参数正确."""
result = build_sidechain_mix_filter(release=0.8)
assert "release=0.800" in result
def test_knee_param(self):
"""knee=6 参数."""
result = build_sidechain_mix_filter()
assert "knee=6" in result
def test_attack_and_release(self):
result = build_sidechain_mix_filter(attack=0.01, release=0.3)
assert "attack=0.010" in result
assert "release=0.300" in result
def test_contains_amix(self):
"""包含 amix 混音."""
result = build_sidechain_mix_filter()
assert "amix=inputs=2" in result
assert "duration=first" in result
def test_volume_compensation(self):
"""volume=1.5 轻微补偿."""
def test_contains_volume_compensation(self):
result = build_sidechain_mix_filter()
assert "volume=1.5" in result
def test_bgmc_comp_label(self):
"""包含 [bgm_comp] 中间标签."""
def test_output_label(self):
result = build_sidechain_mix_filter()
assert "[final]" in result
def test_bgm_comp_label(self):
result = build_sidechain_mix_filter()
assert "[bgm_comp]" in result
# ─────────────────────────────────────────────────────────────────────────────
# normalize_bgm_config 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── normalize_bgm_config ────────────────────────────────────────────────────
class TestNormalizeBGMConfig:
"""配置规范化测试."""
def test_empty_dict_defaults(self):
"""空字典返回默认值."""
class TestNormalizeBgmConfig:
def test_default_values(self):
result = normalize_bgm_config({})
assert result["volume"] == 0.3
assert result["fade_in"] == 0.0
@@ -426,233 +301,184 @@ class TestNormalizeBGMConfig:
assert result["loop_enabled"] is True
assert result["sidechain_enabled"] is False
assert result["sidechain_ratio"] == 0.3
assert result["sidechain_attack"] == 0.02
assert result["sidechain_release"] == 0.5
assert result["sidechain_threshold"] == -25.0
def test_volume_clamped(self):
"""音量钳制."""
result = normalize_bgm_config({"volume": 1.5})
result = normalize_bgm_config({"volume": 2.0})
assert result["volume"] == 1.0
result2 = normalize_bgm_config({"volume": -0.5})
assert result2["volume"] == 0.0
result = normalize_bgm_config({"volume": -1.0})
assert result["volume"] == 0.0
def test_fade_in_negative(self):
"""淡入为负钳制到 0."""
result = normalize_bgm_config({"fade_in": -1})
def test_fade_in_clamped_to_zero(self):
result = normalize_bgm_config({"fade_in": -5})
assert result["fade_in"] == 0.0
def test_fade_out_negative(self):
"""淡出为负钳制到 0."""
result = normalize_bgm_config({"fade_out": -1})
def test_fade_out_clamped_to_zero(self):
result = normalize_bgm_config({"fade_out": -5})
assert result["fade_out"] == 0.0
def test_sidechain_ratio_clamped(self):
"""sidechain_ratio 钳制."""
result = normalize_bgm_config({"sidechain_ratio": 1.5})
assert result["sidechain_ratio"] == 1.0
result2 = normalize_bgm_config({"sidechain_ratio": -0.1})
assert result2["sidechain_ratio"] == 0.0
def test_loop_enabled_bool_conversion(self):
assert normalize_bgm_config({"loop_enabled": True})["loop_enabled"] is True
assert normalize_bgm_config({"loop_enabled": False})["loop_enabled"] is False
assert normalize_bgm_config({"loop_enabled": 1})["loop_enabled"] is True
assert normalize_bgm_config({"loop_enabled": 0})["loop_enabled"] is False
def test_sidechain_attack_min(self):
"""attack 最小值 0.001."""
def test_sidechain_ratio_clamped(self):
result = normalize_bgm_config({"sidechain_ratio": 2.0})
assert result["sidechain_ratio"] == 1.0
result = normalize_bgm_config({"sidechain_ratio": -1.0})
assert result["sidechain_ratio"] == 0.0
def test_sidechain_attack_minimum(self):
result = normalize_bgm_config({"sidechain_attack": 0})
assert result["sidechain_attack"] == 0.001
def test_sidechain_release_min(self):
"""release 最小值 0.01."""
def test_sidechain_release_minimum(self):
result = normalize_bgm_config({"sidechain_release": 0})
assert result["sidechain_release"] == 0.01
def test_sidechain_threshold_pass_through(self):
result = normalize_bgm_config({"sidechain_threshold": -40.0})
assert result["sidechain_threshold"] == -40.0
def test_string_values_converted(self):
"""字符串数值被转换."""
result = normalize_bgm_config(
{
"volume": "0.5",
"fade_in": "2.0",
"fade_in": "1.0",
"sidechain_ratio": "0.7",
}
)
assert result["volume"] == 0.5
assert result["fade_in"] == 2.0
def test_loop_enabled_truthy(self):
"""loop_enabled 真值转换."""
result = normalize_bgm_config({"loop_enabled": 1})
assert result["loop_enabled"] is True
result2 = normalize_bgm_config({"loop_enabled": 0})
assert result2["loop_enabled"] is False
def test_preserves_unknown_keys(self):
"""未知 key 不保留."""
result = normalize_bgm_config({"unknown_key": "value", "volume": 0.5})
assert "unknown_key" not in result
assert result["volume"] == 0.5
assert result["fade_in"] == 1.0
assert result["sidechain_ratio"] == 0.7
# ─────────────────────────────────────────────────────────────────────────────
# validate_bgm_config 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── validate_bgm_config ─────────────────────────────────────────────────────
class TestValidateBGMConfig:
"""配置验证测试."""
class TestValidateBgmConfig:
def test_valid_config(self):
"""合法配置."""
ok, errors = validate_bgm_config(
{
"volume": 0.5,
"fade_in": 1.0,
"fade_out": 2.0,
"sidechain_ratio": 0.3,
}
)
assert ok is True
assert len(errors) == 0
valid, errors = validate_bgm_config({"volume": 0.3})
assert valid is True
assert errors == []
def test_volume_not_number(self):
"""volume 不是数字."""
ok, errors = validate_bgm_config({"volume": "high"})
assert ok is False
def test_invalid_volume_type(self):
valid, errors = validate_bgm_config({"volume": "abc"})
assert valid is False
assert any("volume" in e for e in errors)
def test_volume_out_of_range(self):
"""volume 超出范围."""
ok, errors = validate_bgm_config({"volume": 1.5})
assert ok is False
valid, errors = validate_bgm_config({"volume": -0.1})
assert valid is False
assert any("volume" in e for e in errors)
valid, errors = validate_bgm_config({"volume": 1.1})
assert valid is False
assert any("volume" in e for e in errors)
def test_fade_in_negative(self):
"""fade_in 为负."""
ok, errors = validate_bgm_config({"fade_in": -1})
assert ok is False
def test_volume_at_boundaries(self):
assert validate_bgm_config({"volume": 0})[0] is True
assert validate_bgm_config({"volume": 1})[0] is True
def test_invalid_fade_in_type(self):
valid, errors = validate_bgm_config({"fade_in": "abc"})
assert valid is False
assert any("fade_in" in e for e in errors)
def test_fade_out_negative(self):
"""fade_out 为负."""
ok, errors = validate_bgm_config({"fade_out": -1})
assert ok is False
def test_negative_fade_in(self):
valid, errors = validate_bgm_config({"fade_in": -1})
assert valid is False
assert any("fade_in" in e for e in errors)
def test_invalid_fade_out_type(self):
valid, errors = validate_bgm_config({"fade_out": "abc"})
assert valid is False
assert any("fade_out" in e for e in errors)
def test_negative_fade_out(self):
valid, errors = validate_bgm_config({"fade_out": -1})
assert valid is False
assert any("fade_out" in e for e in errors)
def test_invalid_sidechain_ratio_type(self):
valid, errors = validate_bgm_config({"sidechain_ratio": "abc"})
assert valid is False
assert any("sidechain_ratio" in e for e in errors)
def test_sidechain_ratio_out_of_range(self):
"""sidechain_ratio 超出范围."""
ok, errors = validate_bgm_config({"sidechain_ratio": 2.0})
assert ok is False
valid, errors = validate_bgm_config({"sidechain_ratio": -0.1})
assert valid is False
assert any("sidechain_ratio" in e for e in errors)
valid, errors = validate_bgm_config({"sidechain_ratio": 1.1})
assert valid is False
assert any("sidechain_ratio" in e for e in errors)
def test_multiple_errors(self):
"""多个错误同时报告."""
ok, errors = validate_bgm_config(
valid, errors = validate_bgm_config(
{
"volume": 2.0,
"volume": "bad",
"fade_in": -1,
"sidechain_ratio": -0.5,
"sidechain_ratio": 2.0,
}
)
assert ok is False
assert valid is False
assert len(errors) >= 3
def test_empty_config_valid(self):
"""空配置(全用默认值)视为合法."""
ok, errors = validate_bgm_config({})
assert ok is True
assert len(errors) == 0
# ─────────────────────────────────────────────────────────────────────────────
# calculate_fade_out_start 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── calculate_fade_out_start ────────────────────────────────────────────────
class TestCalculateFadeOutStart:
"""淡出开始时间计算测试."""
def test_normal_case(self):
"""正常情况."""
assert calculate_fade_out_start(100, 3) == pytest.approx(97.0)
assert calculate_fade_out_start(60, 2) == 58.0
def test_zero_fade_out(self):
"""淡出时长为 0,返回 None."""
assert calculate_fade_out_start(100, 0) is None
assert calculate_fade_out_start(60, 0) is None
def test_negative_fade_out(self):
"""淡出时长为负,返回 None."""
assert calculate_fade_out_start(100, -1) is None
def test_zero_duration(self):
"""总时长为 0,返回 None."""
assert calculate_fade_out_start(0, 3) is None
def test_fade_out_longer_than_duration(self):
"""淡出超过总时长,返回 None."""
assert calculate_fade_out_start(10, 20) is None
def test_fade_out_equal_to_duration(self):
"""淡出等于总时长,返回 None."""
assert calculate_fade_out_start(10, 10) is None
# ─────────────────────────────────────────────────────────────────────────────
# estimate_bgm_processing_duration 测试
# ─────────────────────────────────────────────────────────────────────────────
class TestEstimateBGMProcessingDuration:
"""BGM 处理时长估算测试."""
def test_normal_case_with_loop(self):
"""正常循环情况,输出目标时长."""
assert estimate_bgm_processing_duration(10, 100, True) == 100
def test_bgm_longer_no_loop(self):
"""BGM 够长,不循环,截断到目标时长."""
assert estimate_bgm_processing_duration(200, 100, False) == 100
def test_bgm_shorter_no_loop(self):
"""BGM 短但不循环,仍然截断到目标时长(实际会更短,但 atrim 会截断)."""
assert estimate_bgm_processing_duration(10, 100, False) == 100
assert calculate_fade_out_start(60, -1) is None
def test_zero_target(self):
"""目标时长为 0,兜底 5 秒."""
assert estimate_bgm_processing_duration(10, 0, True) == 5.0
assert calculate_fade_out_start(0, 2) is None
def test_negative_target(self):
"""目标时长为负,兜底 5 秒."""
assert estimate_bgm_processing_duration(10, -5, True) == 5.0
assert calculate_fade_out_start(-5, 2) is None
def test_fade_longer_than_target(self):
assert calculate_fade_out_start(10, 20) is None
def test_fade_equal_to_target(self):
assert calculate_fade_out_start(10, 10) is None
def test_float_values(self):
assert calculate_fade_out_start(60.5, 2.5) == 58.0
# ─────────────────────────────────────────────────────────────────────────────
# BGMPureConfig 测试
# ─────────────────────────────────────────────────────────────────────────────
# ── estimate_bgm_processing_duration ────────────────────────────────────────
class TestBGMPureConfig:
"""BGMPureConfig 数据类测试."""
class TestEstimateBgmProcessingDuration:
def test_bgm_longer_no_loop(self):
assert estimate_bgm_processing_duration(100, 60, loop_enabled=False) == 60.0
def test_default_values(self):
"""默认值正确."""
config = BGMPureConfig()
assert config.volume == 0.3
assert config.fade_in == 0.0
assert config.fade_out == 0.0
assert config.loop_enabled is True
assert config.sidechain_enabled is False
assert config.sidechain_ratio == 0.3
assert config.sidechain_attack == 0.02
assert config.sidechain_release == 0.5
assert config.sidechain_threshold == -25.0
def test_bgm_longer_with_loop(self):
# 够长但允许循环,仍然截断到target
assert estimate_bgm_processing_duration(100, 60, loop_enabled=True) == 60.0
def test_custom_values(self):
"""自定义值."""
config = BGMPureConfig(
volume=0.7,
fade_in=1.0,
fade_out=2.0,
loop_enabled=False,
sidechain_enabled=True,
sidechain_ratio=0.5,
sidechain_attack=0.05,
sidechain_release=0.8,
sidechain_threshold=-30.0,
)
assert config.volume == 0.7
assert config.loop_enabled is False
assert config.sidechain_enabled is True
assert config.sidechain_threshold == -30.0
def test_bgm_shorter_with_loop(self):
assert estimate_bgm_processing_duration(10, 60, loop_enabled=True) == 60.0
def test_bgm_shorter_no_loop(self):
# 需要循环但不允许 → 截断到target
assert estimate_bgm_processing_duration(10, 60, loop_enabled=False) == 60.0
def test_zero_target_fallback(self):
assert estimate_bgm_processing_duration(100, 0) == 5.0
def test_negative_target_fallback(self):
assert estimate_bgm_processing_duration(100, -5) == 5.0
def test_equal_duration(self):
assert estimate_bgm_processing_duration(60, 60) == 60.0
File diff suppressed because it is too large Load Diff
+79 -288
View File
@@ -1,9 +1,6 @@
"""通用分页器单元测试."""
from __future__ import annotations
"""pagination 单元测试."""
import pytest
from pydantic import ValidationError
from packages.application.common.pagination import (
PaginatedResponse,
@@ -12,377 +9,171 @@ from packages.application.common.pagination import (
paginate,
)
# ── PaginationParams ────────────────────────────────────────────────────────
class TestPaginationParams:
"""PaginationParams 测试"""
def test_default_values(self):
"""默认值正确"""
params = PaginationParams()
assert params.page == 1
assert params.page_size == 20
def test_offset_first_page(self):
"""第一页 offset 为 0"""
def test_custom_values(self):
params = PaginationParams(page=3, page_size=50)
assert params.page == 3
assert params.page_size == 50
def test_offset_calculation(self):
params = PaginationParams(page=1, page_size=20)
assert params.offset == 0
def test_offset_second_page(self):
"""第二页 offset 计算正确"""
params = PaginationParams(page=2, page_size=20)
assert params.offset == 20
params = PaginationParams(page=3, page_size=20)
assert params.offset == 40
def test_offset_custom_page_size(self):
"""自定义 page_size 的 offset"""
params = PaginationParams(page=3, page_size=10)
assert params.offset == 20
params = PaginationParams(page=10, page_size=50)
assert params.offset == 450
def test_limit_equals_page_size(self):
"""limit 等于 page_size"""
params = PaginationParams(page_size=50)
assert params.limit == 50
params = PaginationParams(page_size=30)
assert params.limit == 30
def test_page_must_be_at_least_1(self):
"""page 不能小于 1"""
with pytest.raises(ValidationError):
with pytest.raises(ValueError):
PaginationParams(page=0)
def test_page_negative_raises(self):
"""page 不能为负数"""
with pytest.raises(ValidationError):
PaginationParams(page=-1)
def test_page_size_must_be_at_least_1(self):
"""page_size 不能小于 1"""
with pytest.raises(ValidationError):
with pytest.raises(ValueError):
PaginationParams(page_size=0)
def test_page_size_max_100(self):
"""page_size 最大 100"""
with pytest.raises(ValidationError):
with pytest.raises(ValueError):
PaginationParams(page_size=101)
def test_page_size_100_is_valid(self):
"""page_size=100 是合法的"""
params = PaginationParams(page_size=100)
assert params.page_size == 100
# ── PaginationMeta ──────────────────────────────────────────────────────────
class TestPaginationMeta:
"""PaginationMeta 测试"""
def test_from_params_first_page(self):
"""第一页元数据"""
params = PaginationParams(page=1, page_size=10)
meta = PaginationMeta.from_params(params, total=25)
assert meta.page == 1
assert meta.page_size == 10
assert meta.total == 25
assert meta.total_pages == 3
assert meta.total_pages == 3 # ceil(25/10)
assert meta.has_next is True
assert meta.has_prev is False
def test_from_params_last_page(self):
"""最后一页元数据"""
params = PaginationParams(page=3, page_size=10)
meta = PaginationMeta.from_params(params, total=25)
assert meta.page == 3
assert meta.total_pages == 3
assert meta.has_next is False
assert meta.has_prev is True
def test_from_params_middle_page(self):
"""中间页元数据"""
params = PaginationParams(page=2, page_size=10)
meta = PaginationMeta.from_params(params, total=50)
assert meta.page == 2
assert meta.total_pages == 5
meta = PaginationMeta.from_params(params, total=25)
assert meta.has_next is True
assert meta.has_prev is True
def test_from_params_single_page(self):
params = PaginationParams(page=1, page_size=20)
meta = PaginationMeta.from_params(params, total=5)
assert meta.total_pages == 1
assert meta.has_next is False
assert meta.has_prev is False
def test_from_params_zero_total(self):
"""总数为 0 时"""
params = PaginationParams(page=1, page_size=20)
meta = PaginationMeta.from_params(params, total=0)
assert meta.total == 0
assert meta.total_pages == 0
assert meta.has_next is False
assert meta.has_prev is False
def test_from_params_exact_multiple(self):
"""总数刚好是 page_size 的整数倍"""
def test_from_params_exact_page_size(self):
params = PaginationParams(page=1, page_size=10)
meta = PaginationMeta.from_params(params, total=30)
assert meta.total_pages == 3
def test_from_params_single_page(self):
"""单页即可放下所有数据"""
params = PaginationParams(page=1, page_size=100)
meta = PaginationMeta.from_params(params, total=50)
meta = PaginationMeta.from_params(params, total=10)
assert meta.total_pages == 1
assert meta.has_next is False
assert meta.has_prev is False
def test_from_params_one_extra(self):
params = PaginationParams(page=1, page_size=10)
meta = PaginationMeta.from_params(params, total=11)
assert meta.total_pages == 2
# ── PaginatedResponse ───────────────────────────────────────────────────────
class TestPaginatedResponse:
"""PaginatedResponse 测试"""
def test_create_success(self):
"""创建分页响应"""
params = PaginationParams(page=1, page_size=10)
data = [1, 2, 3]
response = PaginatedResponse.create(data, params, total=25)
assert response.data == [1, 2, 3]
def test_create_response(self):
params = PaginationParams(page=1, page_size=5)
data = [1, 2, 3, 4, 5]
response = PaginatedResponse.create(data, params, total=15)
assert response.data == data
assert response.pagination.page == 1
assert response.pagination.total == 25
assert response.pagination.total == 15
assert response.pagination.total_pages == 3
def test_create_empty_data(self):
"""空数据分页响应"""
params = PaginationParams(page=1, page_size=20)
response = PaginatedResponse.create([], params, total=0)
assert response.data == []
assert response.pagination.total == 0
assert response.pagination.total_pages == 0
# ── paginate function ───────────────────────────────────────────────────────
class TestPaginateFunction:
"""paginate 函数测试(内存分页)"""
class TestPaginate:
def test_first_page(self):
"""第一页分页"""
items = list(range(30))
params = PaginationParams(page=1, page_size=10)
result = paginate(items, params)
assert result.data == list(range(10))
assert result.pagination.total == 30
assert result.pagination.total_pages == 3
assert result.pagination.has_next is True
assert result.pagination.has_prev is False
def test_second_page(self):
"""第二页分页"""
items = list(range(30))
params = PaginationParams(page=2, page_size=10)
result = paginate(items, params)
assert result.data == list(range(10, 20))
assert result.pagination.page == 2
def test_last_page(self):
"""最后一页分页"""
items = list(range(25))
params = PaginationParams(page=3, page_size=10)
result = paginate(items, params)
assert result.data == list(range(20, 25))
assert len(result.data) == 5
assert result.pagination.has_next is False
assert result.pagination.has_prev is True
def test_empty_list(self):
"""空列表分页"""
params = PaginationParams(page=1, page_size=20)
result = paginate([], params)
assert result.data == []
assert result.pagination.total == 0
assert result.pagination.total_pages == 0
def test_page_beyond_total(self):
"""页码超出总数"""
def test_single_page(self):
items = list(range(5))
params = PaginationParams(page=10, page_size=10)
result = paginate(items, params)
assert result.data == []
assert result.pagination.total == 5
assert result.pagination.total_pages == 1
def test_custom_page_size(self):
"""自定义每页数量"""
items = list(range(100))
params = PaginationParams(page=1, page_size=50)
result = paginate(items, params)
assert len(result.data) == 50
assert result.pagination.total_pages == 2
def test_single_item(self):
"""单条数据"""
items = ["only_one"]
params = PaginationParams(page=1, page_size=10)
result = paginate(items, params)
assert result.data == ["only_one"]
assert result.pagination.total == 1
assert result.pagination.total_pages == 1
def test_generic_type_preserved(self):
"""泛型类型数据正确"""
items = [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]
params = PaginationParams(page=1, page_size=10)
result = paginate(items, params)
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
def test_empty_list(self):
items = []
params = PaginationParams(page=1, page_size=10)
result = paginate(items, params)
assert result.data == []
assert result.pagination.total == 0
assert result.pagination.total_pages == 0
def test_page_beyond_end(self):
items = list(range(5))
params = PaginationParams(page=10, page_size=10)
result = paginate(items, params)
assert result.data == []
assert result.pagination.total == 5
def test_page_size_larger_than_items(self):
items = list(range(5))
params = PaginationParams(page=1, page_size=100)
result = paginate(items, params)
assert result.data == items
assert result.pagination.total_pages == 1
def test_middle_page(self):
items = list(range(100))
params = PaginationParams(page=5, page_size=10)
result = paginate(items, params)
assert result.data == list(range(40, 50))
assert result.pagination.has_next is True
assert result.pagination.has_prev is True
+81 -393
View File
@@ -1,412 +1,100 @@
"""文本分段工具单元测试."""
from __future__ import annotations
import pytest
"""text_splitter 单元测试."""
from packages.application.tts_job.text_splitter import split_text
class TestSplitText:
"""split_text 函数测试"""
def test_empty_string_returns_empty_list(self):
"""空字符串返回空列表"""
def test_empty_text_returns_empty(self):
assert split_text("") == []
def test_whitespace_only_returns_empty_list(self):
"""纯空白字符返回空列表"""
assert split_text(" \n \t ") == []
def test_short_text_returns_single_segment(self):
"""短文本直接返回单段"""
text = "这是一段短文本。"
result = split_text(text, max_chars=500)
assert result == [text]
def test_text_length_equals_max_chars(self):
"""文本长度恰好等于 max_chars 时返回单段"""
text = "a" * 100
result = split_text(text, max_chars=100)
assert len(result) == 1
assert len(result[0]) == 100
def test_splits_on_sentence_boundary(self):
"""在句子边界处分段"""
# 构造长文本,确保超过 max_chars
sentences = ["今天天气真好。我们一起去公园散步吧。", "公园里有很多花。还有很多小朋友在玩耍。"] * 10
text = "".join(sentences)
result = split_text(text, max_chars=200)
assert len(result) >= 2
# 每段都不超过 max_chars
for seg in result:
assert len(seg) <= 200
def test_all_segments_within_max_chars(self):
"""所有分段都不超过 max_chars"""
text = "这是第一句话。这是第二句话。这是第三句话。这是第四句话。这是第五句话。" * 10
result = split_text(text, max_chars=100)
for seg in result:
assert len(seg) <= 100
def test_long_single_sentence_hard_cut(self):
"""超长单句会被硬切"""
text = "a" * 1000 # 没有标点
result = split_text(text, max_chars=200)
assert len(result) > 1
for seg in result:
assert len(seg) <= 200
def test_newline_is_sentence_end(self):
"""换行符作为句子结束符"""
text = "第一行内容\n第二行内容\n第三行内容" * 10
result = split_text(text, max_chars=50)
assert len(result) > 1
for seg in result:
assert len(seg) <= 50
def test_chinese_punctuation(self):
"""中文标点(。!?;)作为句子结束符"""
text = "你好!今天吃什么?我吃米饭;你呢?我也吃米饭。" * 10
result = split_text(text, max_chars=80)
for seg in result:
assert len(seg) <= 80
def test_english_punctuation(self):
"""英文标点(.!?;)作为句子结束符"""
text = "Hello! How are you? I'm fine; thank you. Good bye." * 10
result = split_text(text, max_chars=80)
for seg in result:
assert len(seg) <= 80
def test_merged_short_segments(self):
"""过短的段落会被合并"""
# 构造很多短句
text = "你好。再见。谢谢。抱歉。好的。不行。可以。去吧。" * 5 # 每句3-4字
result = split_text(text, max_chars=100)
# 合并后段数应该比单纯按句切的少
assert len(result) < len(text) // 3 # 粗略估计
for seg in result:
assert len(seg) <= 100
def test_preserves_content(self):
"""分段后内容总和与原文基本一致(忽略strip的空白)"""
text = "这是测试文本。包含多个句子。用来验证分段正确性。" * 5
result = split_text(text, max_chars=50)
# 合并所有分段,去掉空白后应该与原文去掉空白后基本一致
combined = "".join(result).replace(" ", "")
original = text.strip().replace(" ", "")
assert combined == original
def test_custom_max_chars(self):
"""支持自定义 max_chars"""
text = "测试" * 100 # 200字
result_50 = split_text(text, max_chars=50)
result_100 = split_text(text, max_chars=100)
# max_chars 越小,段数应该越多
assert len(result_50) >= len(result_100)
def test_single_char_text(self):
"""单字符文本"""
assert split_text("好", max_chars=10) == ["好"]
def test_text_with_only_punctuation(self):
"""纯标点文本"""
text = "。。。。。。。。。。" # 10个句号
result = split_text(text, max_chars=5)
assert len(result) >= 1
for seg in result:
assert len(seg) <= 5
def test_mixed_content(self):
"""中英文混合内容"""
text = "今天的天气是 sunny and warm。我们去了 park 玩。真的很开心!" * 5
result = split_text(text, max_chars=80)
for seg in result:
assert len(seg) <= 80
# ── 短文本与空文本补充 ──────────────────────────────────────────────────────
class TestSplitTextEmptyAndShort:
"""空文本与短文本补充场景."""
def test_whitespace_only_returns_empty(self):
"""纯空白文本返回空列表."""
def test_whitespace_only(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)
def test_short_text_single_segment(self):
text = "你好世界。"
result = split_text(text, max_chars=500)
assert len(result) == 1
assert result[0] == text
def test_one_over_max_chars_splits(self):
"""超过 max_chars 1 个字符就会分割."""
text = "a" * 101
def test_exact_max_chars(self):
text = "a" * 500
result = split_text(text, max_chars=500)
assert len(result) == 1
assert len(result[0]) == 500
def test_splits_on_sentence_boundary(self):
# 两个长句子,各300字左右,超过50字阈值
sent1 = "你" * 300 + "。"
sent2 = "我" * 300 + "。"
text = sent1 + sent2
result = split_text(text, max_chars=500)
assert len(result) == 2
assert result[0] == sent1
assert result[1] == sent2
def test_long_sentence_hard_cut(self):
# 一个超长句子,没有句末标点,会被硬切
text = "长" * 800
result = split_text(text, max_chars=500)
assert len(result) >= 2
assert all(len(seg) <= 500 for seg in result)
# 合起来应该等于原文本
assert "".join(result) == text
def test_short_segments_merged(self):
# 多个短句应该被合并
sentences = [f"第{i}句。" for i in range(10)]
text = "".join(sentences)
result = split_text(text, max_chars=200)
# 每句5字左右,10句才50字,应该合并成1段
assert len(result) < 10
assert len(result[0]) <= 200
def test_preserves_content(self):
text = "今天天气真好。我们去公园玩吧!你觉得怎么样?好的,走吧。"
result = split_text(text, max_chars=20)
# 合并后内容应一致
assert "".join(result) == text
def test_multiple_punctuation_types(self):
# 构造足够长的文本触发分段
text = "第一" * 30 + "。" + "第二" * 30 + "!" + "第三" * 30 + "?" + "第四" * 30 + ";"
result = split_text(text, max_chars=100)
assert len(result) >= 2
assert "".join(result) == text
def test_none_raises(self):
"""None 输入抛 AttributeError(strip 失败)."""
with pytest.raises(AttributeError):
split_text(None)
def test_custom_max_chars(self):
text = "a" * 100 + "。" + "b" * 100 + "。"
result = split_text(text, max_chars=150)
assert len(result) == 2
assert "a" in result[0]
assert "b" in result[1]
# ── 句子边界分段补充 ──────────────────────────────────────────────────────
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)
def test_newline_as_sentence_end(self):
text = "第一段\n第二段\n第三段"
result = split_text(text, max_chars=50)
assert len(result) >= 1
assert "".join(result) == text.strip()
def test_custom_small_max_chars(self):
"""很小的 max_chars."""
text = "一二三四五六七八九十。" * 5
def test_minimum_segment_length(self):
# 句子太短(<50字)不会立即分段
text = "短句一。短句二。短句三。"
result = split_text(text, max_chars=200)
assert len(result) == 1
def test_trailing_content_added(self):
# 最后一段不完整的句子也要加上
text = "完整的句子。剩余内容"
result = split_text(text, max_chars=50)
assert "".join(result) == text
def test_no_empty_segments(self):
text = "。。。。。" # 全是标点
result = split_text(text, max_chars=2)
assert all(len(seg) > 0 for seg in result)
def test_chinese_and_english_mixed(self):
text = "Hello世界。这是测试Test文本。Mixed混合。"
result = split_text(text, max_chars=20)
assert len(result) > 1
for seg in result:
assert len(seg) <= 20
# ── 更多边界场景补充 ─────────────────────────────────────────────────────────
class TestSplitTextMoreEdgeCases:
"""更多边界场景补充"""
def test_max_chars_one(self):
"""max_chars=1 每个字符一段"""
text = "一二三四五"
result = split_text(text, max_chars=1)
assert len(result) == 5
for seg in result:
assert len(seg) == 1
def test_consecutive_newlines(self):
"""连续多个换行符"""
text = "第一段\n\n\n第二段\n\n第三段"
result = split_text(text, max_chars=100)
# 合并后应该是一段(内容不长且合并逻辑会被合并)
assert len(result) >= 1
assert "第一段" in result[0]
for seg in result:
assert len(seg) <= 100
def test_only_newlines_only(self):
"""只有换行符(纯空白被strip掉返回空"""
assert split_text("\n\n\n\n") == []
def test_leading_trailing_whitespace(self):
"""首尾空白被去除"""
text = " 你好世界。 "
result = split_text(text, max_chars=100)
assert result == ["你好世界。"]
def test_very_long_single_sentence_many_segments(self):
"""超长单句被切成很多段"""
text = "字" * 1000
result = split_text(text, max_chars=100)
assert len(result) == 10
for seg in result:
assert len(seg) == 100
def test_mixed_punctuation_types(self):
"""全角半角标点混合"""
text = "你好!再见。谢谢?抱歉;好的"
result = split_text(text, max_chars=200)
assert len(result) == 1
def test_last_segment_short_merged_to_previous(self):
"""尾部极短段被合并到前一段"""
# 构造第一段接近max_chars,结尾有个短句尾巴
long_part = "一二三四五六七八九十" * 9 + "。" # ~90字
tail = "完" # 1字
text = long_part + tail
result = split_text(text, max_chars=100)
# 尾巴应该被合并
combined = "".join(result)
assert combined == text.strip()
assert len(result) <= 2
def test_all_short_sentences_merged_into_one(self):
"""大量短句全部合并成一段"""
sentences = ["你好。", "我好。", "他好。", "大家好。", "才是真的好。"]
text = "".join(sentences)
result = split_text(text, max_chars=200)
assert len(result) == 1
def test_punctuation_only_long(self):
"""很长的纯标点文本"""
text = "。" * 200
result = split_text(text, max_chars=50)
assert len(result) >= 4
for seg in result:
assert len(seg) <= 50
def test_tab_not_sentence_end(self):
"""制表符不是句子结束符"""
text = "这是一段\t包含制表符的文本内容" + "字" * 100
result = split_text(text, max_chars=50)
# 制表符不在句子结束符集合中,不会触发分段
# 制表符会保留在分段内容中
has_tab = any("\t" in seg for seg in result)
assert has_tab
assert len(result) >= 2
assert "".join(result) == text