xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases 7 Wiki Activity

fix: black format for 6 test files

Browse Source
This commit is contained in:
CI Bot
2026-07-21 00:11:09 +08:00
parent 0c84ec2cf6
commit ea7bef2d5e
6 changed files with 183 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_duplication_domain.py
+38 -54
View File
@@ -116,18 +116,24 @@ class TestDuplicateSegmentCreate:
def test_similarity_boundary_zero(self):
seg = DuplicateSegment.create(
source_start=0, source_end=10,
matched_video_id="v1", matched_video_name="n1",
matched_start=0, matched_end=10,
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=0,
)
assert seg.similarity == 0
def test_similarity_boundary_100(self):
seg = DuplicateSegment.create(
source_start=0, source_end=10,
matched_video_id="v1", matched_video_name="n1",
matched_start=0, matched_end=10,
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=100,
)
assert seg.similarity == 100
@@ -218,24 +224,23 @@ class TestDuplicationRecordLifecycle:
"""生命周期状态转换测试"""
def test_mark_processing(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
old_updated = record.updated_at
record.mark_processing()
assert record.status == "processing"
assert record.updated_at >= old_updated
def test_mark_completed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
segments = [
DuplicateSegment.create(
source_start=0, source_end=10,
matched_video_id="v1", matched_video_name="n1",
matched_start=0, matched_end=10,
source_start=0,
source_end=10,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=10,
similarity=90,
)
]
@@ -251,9 +256,7 @@ class TestDuplicationRecordLifecycle:
assert record.error_message == ""
def test_mark_completed_zero_rate(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=0.0, duplicate_count=0, segments=[])
assert record.status == "completed"
assert record.duplicate_rate == 0.0
@@ -261,30 +264,22 @@ class TestDuplicationRecordLifecycle:
assert record.segments == []
def test_mark_completed_100_rate(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=100.0, duplicate_count=5, segments=[])
assert record.duplicate_rate == 100.0
def test_mark_completed_invalid_rate_negative(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=-1, duplicate_count=0, segments=[])
def test_mark_completed_invalid_rate_over_100(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=101, duplicate_count=0, segments=[])
def test_mark_failed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
record.mark_failed("网络超时")
assert record.status == "failed"
@@ -292,9 +287,7 @@ class TestDuplicationRecordLifecycle:
assert record.duplicate_rate is None
def test_mark_failed_from_pending(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_failed("文件损坏")
assert record.status == "failed"
assert record.error_message == "文件损坏"
@@ -304,42 +297,35 @@ class TestDuplicationRecordRetry:
"""重试逻辑测试"""
def test_can_retry_failed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_failed("error")
assert record.can_retry() is True
def test_cannot_retry_pending(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
assert record.can_retry() is False
def test_cannot_retry_processing(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
assert record.can_retry() is False
def test_cannot_retry_completed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_completed(duplicate_rate=10, duplicate_count=1, segments=[])
assert record.can_retry() is False
def test_reset_for_retry(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.mark_processing()
segments = [
DuplicateSegment.create(
source_start=0, source_end=5,
matched_video_id="v1", matched_video_name="n1",
matched_start=0, matched_end=5,
source_start=0,
source_end=5,
matched_video_id="v1",
matched_video_name="n1",
matched_start=0,
matched_end=5,
similarity=80,
)
]
@@ -359,9 +345,7 @@ class TestDuplicationRecordRetry:
def test_reset_for_retry_from_pending(self):
"""即使从 pending 也能重置（调用方负责判断 can_retry）"""
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k"
)
record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
record.reset_for_retry()
assert record.status == "pending"
assert record.duplicate_count == 0
tests/unit/test_jwt_service.py
-1
View File
@@ -245,7 +245,6 @@ class TestJWTService:
assert access != refresh
class TestJWTHandler:
"""JWT Handler 委托层测试"""
tests/unit/test_password_hasher.py
-1
View File
@@ -172,7 +172,6 @@ class TestPasswordValidator:
assert error is None
class TestPasswordHandler:
"""Password Handler 委托层测试"""
tests/unit/test_subtitle_domain.py
+125 -83
View File
@@ -77,19 +77,23 @@ class TestSubtitleTimelineBasics:
assert tl.total_duration == 0.0
def test_segment_count(self):
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="a", start=0, end=1),
SubtitleSegment(text="b", start=1, end=2),
SubtitleSegment(text="c", start=2, end=3),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="a", start=0, end=1),
SubtitleSegment(text="b", start=1, end=2),
SubtitleSegment(text="c", start=2, end=3),
]
)
assert tl.segment_count == 3
def test_total_chars(self):
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你好", start=0, end=1),
SubtitleSegment(text="世界", start=1, end=2),
SubtitleSegment(text="abcde", start=2, end=3),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0, end=1),
SubtitleSegment(text="世界", start=1, end=2),
SubtitleSegment(text="abcde", start=2, end=3),
]
)
assert tl.total_chars == 9
def test_custom_language(self):
@@ -106,9 +110,11 @@ class TestMergeShortSegments:
def test_single_segment_no_merge(self):
"""单个片段不需要合并"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="a", start=0, end=1),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="a", start=0, end=1),
]
)
result = tl.merge_short_segments(min_chars=8)
assert result.segment_count == 1
assert result.segments[0].text == "a"
@@ -121,12 +127,14 @@ class TestMergeShortSegments:
def test_all_short_segments_merge_into_one(self):
"""所有短片段合并成一个"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你", start=0, end=0.5),
SubtitleSegment(text="好", start=0.5, end=1.0),
SubtitleSegment(text="世", start=1.0, end=1.5),
SubtitleSegment(text="界", start=1.5, end=2.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你", start=0, end=0.5),
SubtitleSegment(text="好", start=0.5, end=1.0),
SubtitleSegment(text="世", start=1.0, end=1.5),
SubtitleSegment(text="界", start=1.5, end=2.0),
]
)
result = tl.merge_short_segments(min_chars=8)
assert result.segment_count == 1
assert result.segments[0].text == "你好世界"
@@ -135,10 +143,12 @@ class TestMergeShortSegments:
def test_merge_short_segments_preserves_timing(self):
"""合并后时间轴正确"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你好", start=1.0, end=2.0),
SubtitleSegment(text="世界", start=2.0, end=3.5),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=1.0, end=2.0),
SubtitleSegment(text="世界", start=2.0, end=3.5),
]
)
result = tl.merge_short_segments(min_chars=10)
assert result.segment_count == 1
assert result.segments[0].start == 1.0
@@ -148,10 +158,12 @@ class TestMergeShortSegments:
"""合并后词级信息保留"""
w1 = SubtitleWord(text="你好", start=0.0, end=1.0)
w2 = SubtitleWord(text="世界", start=1.0, end=2.0)
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
]
)
result = tl.merge_short_segments(min_chars=10)
assert len(result.segments[0].words) == 2
assert result.segments[0].words[0].text == "你好"
@@ -159,12 +171,14 @@ class TestMergeShortSegments:
def test_multiple_merged_groups(self):
"""多个合并组 — 短段会和后续段累积到够数才提交"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字，够数，提交
SubtitleSegment(text="九", start=2, end=2.5), # 1字，入buffer
SubtitleSegment(text="十", start=2.5, end=3), # 1字，入buffer（共2字）
SubtitleSegment(text="一二三四五六七八九十", start=3, end=5), # 10字，入buffer后共12字，够数提交
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字，够数，提交
SubtitleSegment(text="九", start=2, end=2.5), # 1字，入buffer
SubtitleSegment(text="十", start=2.5, end=3), # 1字，入buffer（共2字）
SubtitleSegment(text="一二三四五六七八九十", start=3, end=5), # 10字，入buffer后共12字，够数提交
]
)
result = tl.merge_short_segments(min_chars=8)
# 第1段："一二三四五六七八"（8字直接提交）
# 第2段："九十" + "一二三四五六七八九十" 累积到12字一起提交
@@ -174,10 +188,12 @@ class TestMergeShortSegments:
def test_remaining_short_merged_with_last(self):
"""剩余短片段合并到最后一段"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字
SubtitleSegment(text="一二三", start=2, end=3), # 3字，不够
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二三四五六七八", start=0, end=2), # 8字
SubtitleSegment(text="一二三", start=2, end=3), # 3字，不够
]
)
result = tl.merge_short_segments(min_chars=8)
# 最后的3字会合并到上一段（因为 < min_chars）
assert result.segment_count == 1
@@ -185,11 +201,13 @@ class TestMergeShortSegments:
def test_custom_min_chars(self):
"""自定义最小字数 — 累积到够数就提交，剩余短的合并到最后"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="一二", start=0, end=1),
SubtitleSegment(text="三四", start=1, end=2),
SubtitleSegment(text="五六", start=2, end=3),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="一二", start=0, end=1),
SubtitleSegment(text="三四", start=1, end=2),
SubtitleSegment(text="五六", start=2, end=3),
]
)
# min_chars=3：
# "一二"(2字) → 不够
# +"三四"(共4字) → 够了，提交"一二三四"，buffer清空
@@ -228,9 +246,11 @@ class TestSplitLongSegments:
def test_short_segments_no_split(self):
"""短片段不需要拆分"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你好", start=0, end=1),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好", start=0, end=1),
]
)
result = tl.split_long_segments(max_chars=20)
assert result.segment_count == 1
assert result.segments[0].text == "你好"
@@ -238,9 +258,11 @@ class TestSplitLongSegments:
def test_single_long_segment_split_by_punctuation(self):
"""长片段按标点拆分"""
text = "你好世界。今天天气真好，我们出去玩吧！"
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=text, start=0, end=10.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=10)
# 应该被拆成多段
assert result.segment_count > 1
@@ -251,9 +273,11 @@ class TestSplitLongSegments:
def test_split_preserves_total_text(self):
"""拆分后总文本不变"""
text = "你好世界。今天天气真好，我们出去玩吧！明天再见。"
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=text, start=0, end=10.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=8)
merged_text = "".join(s.text for s in result.segments)
assert merged_text == text
@@ -261,9 +285,11 @@ class TestSplitLongSegments:
def test_split_time_proportional(self):
"""拆分后时间按字数比例分配"""
text = "一二三四五六七八九十。" # 11字
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=text, start=0, end=10.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=5)
# 总时长不变
assert result.segments[0].start == 0.0
@@ -279,9 +305,11 @@ class TestSplitLongSegments:
SubtitleWord(text="世界", start=1.0, end=2.0),
SubtitleWord(text="你好吗", start=2.0, end=3.5),
]
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="你好世界。你好吗？", start=0.0, end=3.5, words=words),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="你好世界。你好吗？", start=0.0, end=3.5, words=words),
]
)
result = tl.split_long_segments(max_chars=4)
# 第一段应该有前几个词
assert len(result.segments) >= 2
@@ -290,11 +318,13 @@ class TestSplitLongSegments:
def test_multiple_mixed_segments(self):
"""混合长短片段"""
tl = SubtitleTimeline(segments=[
SubtitleSegment(text="短", start=0, end=1), # 短
SubtitleSegment(text="一二三四五六七八九十一二三四五六七八九十", start=1, end=5), # 长
SubtitleSegment(text="也短", start=5, end=6), # 短
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text="短", start=0, end=1), # 短
SubtitleSegment(text="一二三四五六七八九十一二三四五六七八九十", start=1, end=5), # 长
SubtitleSegment(text="也短", start=5, end=6), # 短
]
)
result = tl.split_long_segments(max_chars=10)
assert result.segment_count >= 3 # 至少3段（中间被拆成多段）
# 第一段还是原来的短的
@@ -305,9 +335,11 @@ class TestSplitLongSegments:
def test_no_punctuation_hard_split(self):
"""没有标点时硬切"""
text = "一二三四五六七八九十一二三四五六七八九十一二三四五"
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=text, start=0, end=10.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=text, start=0, end=10.0),
]
)
result = tl.split_long_segments(max_chars=10)
assert result.segment_count >= 3
for seg in result.segments:
@@ -328,9 +360,11 @@ class TestSplitLongSegments:
def test_does_not_modify_original(self):
"""不修改原时间轴"""
original_text = "一二三四五六七八九十一二三四五六七八九十"
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=original_text, start=0, end=5),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=original_text, start=0, end=5),
]
)
result = tl.split_long_segments(max_chars=8)
assert tl.segment_count == 1
assert tl.segments[0].text == original_text
@@ -401,10 +435,12 @@ class TestMergeSegments:
"""_merge_segments 静态方法测试"""
def test_merge_two_segments(self):
result = SubtitleTimeline._merge_segments([
SubtitleSegment(text="你好", start=0.0, end=1.0),
SubtitleSegment(text="世界", start=1.0, end=2.0),
])
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="你好", start=0.0, end=1.0),
SubtitleSegment(text="世界", start=1.0, end=2.0),
]
)
assert result.text == "你好世界"
assert result.start == 0.0
assert result.end == 2.0
@@ -425,20 +461,24 @@ class TestMergeSegments:
def test_merge_preserves_words(self):
w1 = SubtitleWord(text="你好", start=0.0, end=1.0)
w2 = SubtitleWord(text="世界", start=1.0, end=2.0)
result = SubtitleTimeline._merge_segments([
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
])
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="你好", start=0.0, end=1.0, words=[w1]),
SubtitleSegment(text="世界", start=1.0, end=2.0, words=[w2]),
]
)
assert len(result.words) == 2
assert result.words[0].text == "你好"
assert result.words[1].text == "世界"
def test_merge_non_contiguous_segments(self):
"""合并非连续片段（有间隙）"""
result = SubtitleTimeline._merge_segments([
SubtitleSegment(text="a", start=0.0, end=1.0),
SubtitleSegment(text="b", start=3.0, end=4.0),
])
result = SubtitleTimeline._merge_segments(
[
SubtitleSegment(text="a", start=0.0, end=1.0),
SubtitleSegment(text="b", start=3.0, end=4.0),
]
)
assert result.start == 0.0
assert result.end == 4.0
assert result.text == "ab"
@@ -450,9 +490,11 @@ class TestMergeAndSplitRoundtrip:
def test_split_then_merge_approximate(self):
"""拆分后再合并，总字数和总时长基本一致"""
original_text = "你好世界。今天天气真好，我们出去玩吧！明天见。"
tl = SubtitleTimeline(segments=[
SubtitleSegment(text=original_text, start=0.0, end=10.0),
])
tl = SubtitleTimeline(
segments=[
SubtitleSegment(text=original_text, start=0.0, end=10.0),
]
)
split = tl.split_long_segments(max_chars=5)
merged = split.merge_short_segments(min_chars=50) # 足够大的min_chars让它们都合并
assert merged.segment_count == 1
tests/unit/test_tts_config_domain.py
+19 -15
View File
@@ -40,25 +40,29 @@ class TestTtsConfigParse:
def test_parse_disabled_returns_minimal(self):
"""disabled 时直接返回 enabled=False，忽略其他字段"""
config = TtsConfig.parse({
"enabled": False,
"voice_id": "v123",
"speed": 1.5,
})
config = TtsConfig.parse(
{
"enabled": False,
"voice_id": "v123",
"speed": 1.5,
}
)
assert config.enabled is False
assert config.voice_id == "" # 不保留
def test_parse_enabled_true(self):
config = TtsConfig.parse({
"enabled": True,
"voice_id": "voice_001",
"speed": 1.2,
"pitch": 2.5,
"volume": 0.5,
"text": "你好世界",
"align_mode": "subtitle",
"overlap_mode": "mix",
})
config = TtsConfig.parse(
{
"enabled": True,
"voice_id": "voice_001",
"speed": 1.2,
"pitch": 2.5,
"volume": 0.5,
"text": "你好世界",
"align_mode": "subtitle",
"overlap_mode": "mix",
}
)
assert config.enabled is True
assert config.voice_id == "voice_001"
assert config.speed == 1.2
tests/unit/test_voice_clone_profile_domain.py
+1 -3
View File
@@ -384,8 +384,6 @@ class TestToDict:
assert d["is_retryable"] is True # retry_count=0, max_retries=3
def test_to_dict_includes_metadata(self):
profile = VoiceCloneProfile.create(
user_id="u1", name="test", metadata={"key": "value", "num": 42}
)
profile = VoiceCloneProfile.create(user_id="u1", name="test", metadata={"key": "value", "num": 42})
d = profile.to_dict()
assert d["metadata"] == {"key": "value", "num": 42}