xiaoxia-saas/tests/unit/test_mock_asr_service.py

"""MockASRService 单测 — Mock ASR服务纯逻辑部分."""

from __future__ import annotations

from pathlib import Path

import pytest

from packages.adapters.asr.mock_asr_service import MockASRService
from packages.ports.asr_service import ASRServiceError

# ── Fixtures ────────────────────────────────────────────────────────────────


@pytest.fixture
def service():
    return MockASRService()


# ── transcribe 基础行为 ────────────────────────────────────────────────────────


class TestTranscribeBasic:
    """transcribe 基本行为."""

    def test_audio_not_found_raises(self, service, tmp_path):
        """音频文件不存在时抛 ASRServiceError."""
        with pytest.raises(ASRServiceError):
            service.transcribe(tmp_path / "nonexistent.wav")

    def test_default_mock_text_produces_timeline(self, service, tmp_path):
        """不传 mock_text 时生成默认测试字幕."""
        audio = tmp_path / "test.wav"
        audio.write_bytes(b"fake audio")

        timeline = service.transcribe(audio)

        assert timeline is not None
        assert timeline.language == "zh"
        assert timeline.total_duration > 0
        assert len(timeline.segments) > 0

    def test_custom_mock_text(self, tmp_path):
        """使用自定义 mock_text."""
        text = "你好世界。今天天气真好！"
        service = MockASRService(mock_text=text)
        audio = tmp_path / "test.wav"
        audio.write_bytes(b"fake audio")

        timeline = service.transcribe(audio)

        assert timeline.total_duration > 0
        # 两句话，应该有2个segment
        assert len(timeline.segments) == 2
        assert "你好世界" in timeline.segments[0].text
        assert "今天天气真好" in timeline.segments[1].text

    def test_reads_txt_file(self, tmp_path):
        """音频同目录同名 txt 文件存在时，读取其内容."""
        audio = tmp_path / "voice.wav"
        audio.write_bytes(b"fake audio")
        txt = tmp_path / "voice.txt"
        txt.write_text("这是从文件读取的。内容。", encoding="utf-8")

        service = MockASRService()
        timeline = service.transcribe(audio)

        assert len(timeline.segments) == 2
        assert "从文件读取" in timeline.segments[0].text

    def test_mock_text_takes_priority_over_txt(self, tmp_path):
        """mock_text 参数优先于 txt 文件."""
        audio = tmp_path / "test.wav"
        audio.write_bytes(b"fake audio")
        txt = tmp_path / "test.txt"
        txt.write_text("文件里的文字。", encoding="utf-8")

        service = MockASRService(mock_text="自定义的。优先！")
        timeline = service.transcribe(audio)

        assert "自定义的" in timeline.segments[0].text
        assert "文件" not in timeline.segments[0].text

    def test_custom_language(self, service, tmp_path):
        """可以指定语言."""
        audio = tmp_path / "test.wav"
        audio.write_bytes(b"fake audio")
        timeline = service.transcribe(audio, language="en")
        assert timeline.language == "en"


# ── _text_to_segments 切分逻辑 ──────────────────────────────────────────


class TestTextToSegments:
    """_text_to_segments 文本切分逻辑."""

    def setup_method(self):
        self.service = MockASRService()

    def test_split_by_chinese_period(self):
        """按中文句号切分."""
        text = "第一句。第二句。第三句。"
        segments = self.service._text_to_segments(text, 10.0, with_word_timestamps=False)
        assert len(segments) == 3

    def test_split_by_exclamation(self):
        """按感叹号切分."""
        text = "你好！世界！"
        segments = self.service._text_to_segments(text, 5.0, with_word_timestamps=False)
        assert len(segments) == 2

    def test_split_by_question_mark(self):
        """按问号切分."""
        text = "你好吗？我很好。"
        segments = self.service._text_to_segments(text, 5.0, with_word_timestamps=False)
        assert len(segments) == 2

    def test_mixed_punctuation(self):
        """混合标点."""
        text = "你好！你是谁？我是测试。再见!"
        segments = self.service._text_to_segments(text, 10.0, with_word_timestamps=False)
        assert len(segments) == 4

    def test_empty_text_returns_empty(self):
        """空文本返回空列表."""
        segments = self.service._text_to_segments("", 5.0, with_word_timestamps=False)
        assert segments == []

    def test_no_punctuation_single_segment(self):
        """没有标点时整段作为一个segment."""
        text = "这是一段没有标点的文字"
        segments = self.service._text_to_segments(text, 5.0, with_word_timestamps=False)
        assert len(segments) == 1
        assert segments[0].text == text

    def test_segments_duration_adds_up(self):
        """所有segment时长加起来约等于总时长."""
        text = "第一句。第二句。"
        total = 10.0
        segments = self.service._text_to_segments(text, total, with_word_timestamps=False)
        sum_duration = sum(s.end - s.start for s in segments)
        assert abs(sum_duration - total) < 0.01

    def test_segments_sequential(self):
        """segment按顺序排列，首尾相接."""
        text = "第一句。第二句。第三句。"
        segments = self.service._text_to_segments(text, 9.0, with_word_timestamps=False)

        assert segments[0].start == 0.0
        for i in range(1, len(segments)):
            assert abs(segments[i].start - segments[i - 1].end) < 0.001

    def test_with_word_timestamps(self):
        """带词级时间戳时，每个字一个word."""
        text = "你好世界！"
        segments = self.service._text_to_segments(text, 2.0, with_word_timestamps=True)

        assert len(segments) == 1
        # 4个汉字 + 1个感叹号 = 5个字符
        assert len(segments[0].words) == 5

    def test_word_timestamps_sequential(self):
        """词级时间戳按顺序排列."""
        text = "你好！"
        segments = self.service._text_to_segments(text, 3.0, with_word_timestamps=True)

        words = segments[0].words
        assert len(words) == 3
        assert words[0].start == 0.0
        # 最后一个词的结束时间约等于 segment 结束时间
        assert abs(words[-1].end - segments[0].end) < 0.01

    def test_without_word_timestamps(self):
        """不带词级时间戳时，words为空."""
        text = "你好世界。"
        segments = self.service._text_to_segments(text, 2.0, with_word_timestamps=False)

        assert len(segments) == 1
        assert segments[0].words == []

    def test_duration_proportional_to_length(self):
        """长句子占时长，短句子占时短."""
        text = "短。很长很长很长很长的句子。"
        segments = self.service._text_to_segments(text, 10.0, with_word_timestamps=False)

        # 第一句1个字，第二句9个字（包括标点)
        # 第一句时长应该比第二句短
        dur0 = segments[0].end - segments[0].start
        dur1 = segments[1].end - segments[1].start
        assert dur1 > dur0  # 第二句更长