test(14-05): fix semantic segmentation toggle tests
- Fix patch path for SimpleAIClient (deep_analyzer not semantic_engine) - Remove config parameter from SRTProcessor initialization - Add proper assertions for cache enabled/disabled states - Import Mock from unittest.mock - Remove duplicate test functions Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+18
-25
@@ -14,6 +14,7 @@ import pytest
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import List, Optional
|
from typing import List, Optional
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
@@ -504,74 +505,66 @@ class TestSemanticSegmentationToggle:
|
|||||||
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
|
with patch('kb_workflow.deep_analyzer.SimpleAIClient') as mock_client:
|
||||||
mock_client.return_value = Mock(api_key="test_key")
|
mock_client.return_value = Mock(api_key="test_key")
|
||||||
|
|
||||||
processor = SRTProcessor(
|
processor = SRTProcessor(
|
||||||
config={},
|
|
||||||
enable_semantic_segmentation=True
|
enable_semantic_segmentation=True
|
||||||
)
|
)
|
||||||
|
|
||||||
# 验证创建了 SemanticSegmentationEngine
|
# 验证创建了 SemanticSegmentationEngine
|
||||||
assert hasattr(processor, 'semantic_engine')
|
from kb_workflow.srt.semantic_engine import SemanticSegmentationEngine
|
||||||
# 注意:当前实现可能还没有这些属性,这个测试在实现后会通过
|
assert isinstance(processor.segmenter, SemanticSegmentationEngine)
|
||||||
# assert processor.semantic_engine is not None
|
# 验证缓存已启用(默认)
|
||||||
# assert processor.segmenter is None
|
assert processor.segmenter.strategy.cache is not None
|
||||||
|
|
||||||
def test_semantic_segmentation_disabled(self, sample_srt_file_for_toggle):
|
def test_semantic_segmentation_disabled(self, sample_srt_file_for_toggle):
|
||||||
"""测试禁用语义分段(默认)"""
|
"""测试禁用语义分段(默认)"""
|
||||||
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
||||||
|
from kb_workflow.srt.segmenter import SemanticSegmenter
|
||||||
|
|
||||||
processor = SRTProcessor(
|
processor = SRTProcessor(
|
||||||
config={},
|
|
||||||
enable_semantic_segmentation=False
|
enable_semantic_segmentation=False
|
||||||
)
|
)
|
||||||
|
|
||||||
# 验证使用规则分段器
|
# 验证使用规则分段器
|
||||||
# 注意:当前实现可能还没有这些属性,这个测试在实现后会通过
|
assert isinstance(processor.segmenter, SemanticSegmenter)
|
||||||
# assert processor.semantic_engine is None
|
|
||||||
# assert processor.segmenter is not None
|
|
||||||
pass
|
|
||||||
|
|
||||||
def test_semantic_segmentation_with_cache(self, sample_srt_file_for_toggle):
|
def test_semantic_segmentation_with_cache(self, sample_srt_file_for_toggle):
|
||||||
"""测试启用语义分段和缓存"""
|
"""测试启用语义分段和缓存"""
|
||||||
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
|
with patch('kb_workflow.deep_analyzer.SimpleAIClient') as mock_client:
|
||||||
mock_client.return_value = Mock(api_key="test_key")
|
mock_client.return_value = Mock(api_key="test_key")
|
||||||
|
|
||||||
processor = SRTProcessor(
|
processor = SRTProcessor(
|
||||||
config={},
|
|
||||||
enable_semantic_segmentation=True,
|
enable_semantic_segmentation=True,
|
||||||
enable_cache=True
|
enable_cache=True
|
||||||
)
|
)
|
||||||
|
|
||||||
# 验证创建了缓存
|
# 验证创建了 SemanticSegmentationEngine 且缓存已启用
|
||||||
# 注意:当前实现可能还没有这些属性,这个测试在实现后会通过
|
from kb_workflow.srt.semantic_engine import SemanticSegmentationEngine
|
||||||
# assert processor.semantic_engine is not None
|
assert isinstance(processor.segmenter, SemanticSegmentationEngine)
|
||||||
# assert processor.semantic_engine.strategy.cache is not None
|
assert processor.segmenter.strategy.cache is not None
|
||||||
pass
|
|
||||||
|
|
||||||
def test_semantic_segmentation_without_cache(self, sample_srt_file_for_toggle):
|
def test_semantic_segmentation_without_cache(self, sample_srt_file_for_toggle):
|
||||||
"""测试启用语义分段但禁用缓存"""
|
"""测试启用语义分段但禁用缓存"""
|
||||||
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
from kb_workflow.pipeline.processors.srt import SRTProcessor
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
with patch('kb_workflow.srt.semantic_engine.SimpleAIClient') as mock_client:
|
with patch('kb_workflow.deep_analyzer.SimpleAIClient') as mock_client:
|
||||||
mock_client.return_value = Mock(api_key="test_key")
|
mock_client.return_value = Mock(api_key="test_key")
|
||||||
|
|
||||||
processor = SRTProcessor(
|
processor = SRTProcessor(
|
||||||
config={},
|
|
||||||
enable_semantic_segmentation=True,
|
enable_semantic_segmentation=True,
|
||||||
enable_cache=False
|
enable_cache=False
|
||||||
)
|
)
|
||||||
|
|
||||||
# 验证没有创建缓存
|
# 验证创建了 SemanticSegmentationEngine 但缓存未启用
|
||||||
# 注意:当前实现可能还没有这些属性,这个测试在实现后会通过
|
from kb_workflow.srt.semantic_engine import SemanticSegmentationEngine
|
||||||
# assert processor.semantic_engine is not None
|
assert isinstance(processor.segmenter, SemanticSegmentationEngine)
|
||||||
# assert processor.semantic_engine.strategy.cache is None
|
assert processor.segmenter.strategy.cache is None
|
||||||
pass
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def sample_srt_file_for_toggle(self, tmp_path):
|
def sample_srt_file_for_toggle(self, tmp_path):
|
||||||
|
|||||||
Reference in New Issue
Block a user