xiaoxia-saas/tests/unit/test_sentence_timings.py

"""Tests for sentence timing functions in lipsync_tts."""

import os
import subprocess
import tempfile
import unittest
from unittest.mock import MagicMock, patch

from apps.api.app.tasks.lipsync_tts import (
    _compute_sentence_timings,
    _estimate_sentence_timings_by_chars,
    _split_script_into_sentences,
)


class TestSplitScriptIntoSentences(unittest.TestCase):
    """Tests for _split_script_into_sentences."""

    def test_empty_string(self):
        self.assertEqual(_split_script_into_sentences(""), [])

    def test_none(self):
        self.assertEqual(_split_script_into_sentences(None), [])

    def test_whitespace_only(self):
        self.assertEqual(_split_script_into_sentences("   \n  "), [])

    def test_single_sentence(self):
        self.assertEqual(_split_script_into_sentences("你好世界。"), ["你好世界"])

    def test_multiple_sentences_chinese(self):
        result = _split_script_into_sentences("第一句。第二句！第三句？")
        self.assertEqual(result, ["第一句", "第二句", "第三句"])

    def test_english_punctuation(self):
        result = _split_script_into_sentences("Hello World! How are you?")
        self.assertEqual(result, ["Hello World", "How are you"])

    def test_semicolons(self):
        result = _split_script_into_sentences("第一部分；第二部分;第三部分")
        self.assertEqual(result, ["第一部分", "第二部分", "第三部分"])

    def test_newlines(self):
        result = _split_script_into_sentences("第一行\n第二行\n第三行")
        self.assertEqual(result, ["第一行", "第二行", "第三行"])

    def test_no_trailing_punctuation(self):
        result = _split_script_into_sentences("没有标点的句子")
        self.assertEqual(result, ["没有标点的句子"])


class TestEstimateSentenceTimingsByChars(unittest.TestCase):
    """Tests for _estimate_sentence_timings_by_chars."""

    def test_empty_sentences(self):
        self.assertEqual(_estimate_sentence_timings_by_chars([], 10.0), [])

    def test_zero_duration(self):
        self.assertEqual(_estimate_sentence_timings_by_chars(["hello"], 0), [])

    def test_negative_duration(self):
        self.assertEqual(_estimate_sentence_timings_by_chars(["hello"], -5.0), [])

    def test_single_sentence(self):
        result = _estimate_sentence_timings_by_chars(["hello"], 10.0)
        self.assertEqual(len(result), 1)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)
        self.assertAlmostEqual(result[0]["end_time"], 10.0)

    def test_two_equal_sentences(self):
        result = _estimate_sentence_timings_by_chars(["你好", "世界"], 10.0)
        self.assertEqual(len(result), 2)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)
        self.assertAlmostEqual(result[0]["end_time"], 5.0)
        self.assertAlmostEqual(result[1]["start_time"], 5.0)
        self.assertAlmostEqual(result[1]["end_time"], 10.0)

    def test_unequal_char_distribution(self):
        result = _estimate_sentence_timings_by_chars(["ABCD", "EF"], 9.0)
        self.assertEqual(len(result), 2)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)
        self.assertAlmostEqual(result[0]["end_time"], 6.0)  # 4/6 * 9 = 6
        self.assertAlmostEqual(result[1]["start_time"], 6.0)
        self.assertAlmostEqual(result[1]["end_time"], 9.0)

    def test_timing_structure(self):
        result = _estimate_sentence_timings_by_chars(["句子一", "句子二"], 6.0)
        for item in result:
            self.assertIn("index", item)
            self.assertIn("text", item)
            self.assertIn("start_time", item)
            self.assertIn("end_time", item)


class TestComputeSentenceTimings(unittest.TestCase):
    """Tests for _compute_sentence_timings."""

    def test_empty_script_returns_empty(self):
        self.assertEqual(_compute_sentence_timings(b"fake_audio", "", 10.0), [])

    def test_none_script_returns_empty(self):
        self.assertEqual(_compute_sentence_timings(b"fake_audio", None, 10.0), [])

    @patch("os.unlink")
    @patch.object(tempfile, "NamedTemporaryFile")
    @patch.object(subprocess, "run")
    def test_silence_detection_insufficient_fallback(self, mock_run, mock_tmpfile, mock_unlink):
        """When silence detection finds too few points, fallback to char estimation."""
        mock_run.return_value = MagicMock(stderr="", returncode=0)
        mock_tmp = MagicMock()
        mock_tmp.name = "/tmp/fake.mp3"
        mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
        mock_tmp.__exit__ = MagicMock(return_value=False)
        mock_tmpfile.return_value = mock_tmp

        result = _compute_sentence_timings(b"fake_audio", "第一句。第二句。第三句。", 10.0)

        # Should fallback to char estimation with 3 sentences
        self.assertEqual(len(result), 3)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)

    @patch("os.unlink")
    @patch.object(tempfile, "NamedTemporaryFile")
    @patch.object(subprocess, "run")
    def test_silence_detection_with_enough_points(self, mock_run, mock_tmpfile, mock_unlink):
        """When silence detection finds enough points, use them for boundaries."""
        mock_run.return_value = MagicMock(
            stderr="[silencedetect] silence_end: 3.5 | silence_duration: 0.4\n"
            "[silencedetect] silence_end: 7.0 | silence_duration: 0.3\n",
            returncode=0,
        )
        mock_tmp = MagicMock()
        mock_tmp.name = "/tmp/fake.mp3"
        mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
        mock_tmp.__exit__ = MagicMock(return_value=False)
        mock_tmpfile.return_value = mock_tmp

        result = _compute_sentence_timings(b"fake_audio", "第一句。第二句。第三句。", 10.0)

        self.assertEqual(len(result), 3)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)
        self.assertAlmostEqual(result[0]["end_time"], 3.5)
        self.assertAlmostEqual(result[1]["start_time"], 3.5)
        self.assertAlmostEqual(result[1]["end_time"], 7.0)
        self.assertAlmostEqual(result[2]["start_time"], 7.0)
        self.assertAlmostEqual(result[2]["end_time"], 10.0)

    @patch("os.unlink")
    @patch.object(tempfile, "NamedTemporaryFile")
    @patch.object(subprocess, "run")
    def test_ffmpeg_exception_fallback(self, mock_run, mock_tmpfile, mock_unlink):
        """When ffmpeg raises an exception, fallback to char estimation."""
        mock_run.side_effect = Exception("ffmpeg not found")
        mock_tmp = MagicMock()
        mock_tmp.name = "/tmp/fake.mp3"
        mock_tmp.__enter__ = MagicMock(return_value=mock_tmp)
        mock_tmp.__exit__ = MagicMock(return_value=False)
        mock_tmpfile.return_value = mock_tmp

        result = _compute_sentence_timings(b"fake_audio", "句子一。句子二。", 6.0)

        # Should fallback to char estimation
        self.assertEqual(len(result), 2)
        self.assertAlmostEqual(result[0]["start_time"], 0.0)
        self.assertAlmostEqual(result[0]["end_time"], 3.0)
        self.assertAlmostEqual(result[1]["start_time"], 3.0)
        self.assertAlmostEqual(result[1]["end_time"], 6.0)


if __name__ == "__main__":
    unittest.main()