From 5cb2b03ea362a0674d411402973d50273581b1a0 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Fri, 24 Jul 2026 14:49:37 +0800
Subject: [PATCH] =?UTF-8?q?test:=20P3-1=20=E7=AC=AC46=E6=B3=A2=E5=8D=95?=
 =?UTF-8?q?=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88api=5Fsettings/worker=5Fset?=
 =?UTF-8?q?tings=20+=20tts=5Fstreaming=E8=A1=A5=E5=85=85=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 新增 test_api_settings.py：SharedSettings/APISettings/WorkerSettings 默认值、JWT校验、CORS解析、UPPER_CASE别名、配置缓存单例，共61个
- 补充 test_tts_streaming.py：边界值(500/501字)、参数传递、分块大小边界、异常处理、分段数量验证，+12个

合计 +73 个测试
---
 tests/unit/test_api_settings.py  | 322 +++++++++++++++++++++++++++++++
 tests/unit/test_tts_streaming.py | 212 ++++++++++++++++++++
 2 files changed, 534 insertions(+)
 create mode 100755 tests/unit/test_api_settings.py

diff --git a/tests/unit/test_api_settings.py b/tests/unit/test_api_settings.py
new file mode 100755
index 000000000..5c25fe58e
--- /dev/null
+++ b/tests/unit/test_api_settings.py
@@ -0,0 +1,322 @@
+"""API Settings 配置单元测试."""
+import os
+
+import pytest
+
+from packages.config.api_settings import APISettings, get_api_settings
+from packages.config.base import SharedSettings, get_cached_settings, reload_settings_cache
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache():
+    """每个测试前清空配置缓存，避免单例污染."""
+    reload_settings_cache()
+    # 设置必要的环境变量，避免 JWT 校验失败
+    os.environ["JWT_SECRET_KEY"] = "test-secret-key-for-unit-tests-only-12345"
+    yield
+    reload_settings_cache()
+    os.environ.pop("JWT_SECRET_KEY", None)
+
+
+class TestSharedSettingsDefaults:
+    """SharedSettings 默认值测试."""
+
+    def test_default_environment(self):
+        s = SharedSettings()
+        assert s.environment == "development"
+
+    def test_default_debug_true(self):
+        s = SharedSettings()
+        assert s.debug is True
+
+    def test_default_auto_create_schema_false(self):
+        s = SharedSettings()
+        assert s.auto_create_schema is False
+
+    def test_default_database_url(self):
+        s = SharedSettings()
+        assert "postgresql" in s.database_url
+        assert "localhost" in s.database_url
+
+    def test_default_database_pool_size(self):
+        s = SharedSettings()
+        assert s.database_pool_size == 20
+
+    def test_default_database_max_overflow(self):
+        s = SharedSettings()
+        assert s.database_max_overflow == 10
+
+    def test_default_redis_url(self):
+        s = SharedSettings()
+        assert s.redis_url.startswith("redis://")
+
+    def test_default_celery_broker_url(self):
+        s = SharedSettings()
+        assert s.celery_broker_url.startswith("redis://")
+
+    def test_default_oss_endpoint(self):
+        s = SharedSettings()
+        assert "aliyuncs.com" in s.oss_endpoint
+
+    def test_default_cosyvoice_settings(self):
+        s = SharedSettings()
+        assert s.cosyvoice_model == "cosyvoice-v3-flash"
+        assert s.cosyvoice_format == "mp3"
+        assert s.cosyvoice_sample_rate == 22050
+
+    def test_default_doubao_settings(self):
+        s = SharedSettings()
+        assert "doubao" in s.doubao_model
+        assert s.doubao_timeout == 30
+        assert s.doubao_max_retries == 2
+
+
+class TestAPISettingsDefaults:
+    """APISettings 默认值测试."""
+
+    def test_default_app_name(self):
+        s = APISettings()
+        assert s.app_name == "xiaoxia-saas"
+
+    def test_default_app_version(self):
+        s = APISettings()
+        assert s.app_version == "0.1.61"
+
+    def test_default_api_host(self):
+        s = APISettings()
+        assert s.api_host == "0.0.0.0"
+
+    def test_default_api_port(self):
+        s = APISettings()
+        assert s.api_port == 8000
+
+    def test_default_jwt_algorithm(self):
+        s = APISettings()
+        assert s.jwt_algorithm == "HS256"
+
+    def test_default_jwt_access_expire(self):
+        s = APISettings()
+        assert s.jwt_access_token_expire_minutes == 30
+
+    def test_default_jwt_refresh_expire(self):
+        s = APISettings()
+        assert s.jwt_refresh_token_expire_days == 30
+
+    def test_default_enable_email_delivery_false(self):
+        s = APISettings()
+        assert s.enable_email_delivery is False
+
+    def test_default_smtp_config(self):
+        s = APISettings()
+        assert s.smtp_host == "smtp.gmail.com"
+        assert s.smtp_port == 587
+        assert s.smtp_use_tls is True
+        assert s.smtp_from_name == "小虾 SaaS"
+
+    def test_default_render_engine(self):
+        s = APISettings()
+        assert s.render_engine == "legacy"
+
+    def test_use_in_memory_db_field_exists(self, monkeypatch):
+        monkeypatch.delenv("USE_IN_MEMORY_DB", raising=False)
+        s = APISettings()
+        assert isinstance(s.use_in_memory_db, bool)
+        assert s.USE_IN_MEMORY_DB == s.use_in_memory_db
+
+    def test_default_enable_redis_sessions_false(self):
+        s = APISettings()
+        assert s.enable_redis_sessions is False
+
+
+class TestJWTSecretValidation:
+    """JWT 密钥校验测试."""
+
+    def test_missing_jwt_secret_raises(self, monkeypatch):
+        monkeypatch.delenv("JWT_SECRET_KEY", raising=False)
+        with pytest.raises(ValueError, match="JWT_SECRET_KEY must be set"):
+            APISettings()
+
+    def test_empty_jwt_secret_raises(self, monkeypatch):
+        monkeypatch.setenv("JWT_SECRET_KEY", "")
+        with pytest.raises(ValueError, match="JWT_SECRET_KEY must be set"):
+            APISettings()
+
+    @pytest.mark.parametrize(
+        "insecure_value",
+        ["your-secret-key-change-in-production", "your-secret-key", "secret", "changeme", "password"],
+    )
+    def test_insecure_jwt_secret_raises(self, monkeypatch, insecure_value):
+        monkeypatch.setenv("JWT_SECRET_KEY", insecure_value)
+        with pytest.raises(ValueError, match="insecure"):
+            APISettings()
+
+    def test_strong_jwt_secret_accepted(self, monkeypatch):
+        monkeypatch.setenv("JWT_SECRET_KEY", "strong-random-secret-key-12345-abcde")
+        s = APISettings()
+        assert s.jwt_secret_key == "strong-random-secret-key-12345-abcde"
+
+
+class TestCorsOrigins:
+    """CORS 配置解析测试."""
+
+    def test_default_cors_origins(self):
+        s = APISettings()
+        origins = s.cors_origins
+        assert isinstance(origins, list)
+        assert len(origins) == 3
+        assert "http://localhost:3000" in origins
+        assert "http://localhost:5173" in origins
+        assert "http://localhost:8000" in origins
+
+    def test_cors_origins_strips_whitespace(self, monkeypatch):
+        monkeypatch.setenv("CORS_ORIGINS_RAW", " http://a.com , http://b.com ")
+        s = APISettings()
+        assert s.cors_origins == ["http://a.com", "http://b.com"]
+
+    def test_cors_origins_empty_string(self, monkeypatch):
+        monkeypatch.setenv("CORS_ORIGINS_RAW", "")
+        s = APISettings()
+        assert s.cors_origins == []
+
+    def test_cors_origins_single_origin(self, monkeypatch):
+        monkeypatch.setenv("CORS_ORIGINS_RAW", "https://api.example.com")
+        s = APISettings()
+        assert s.cors_origins == ["https://api.example.com"]
+
+
+class TestUpperCaseAliases:
+    """向后兼容：UPPER_CASE property 别名测试."""
+
+    def test_app_name_alias(self):
+        s = APISettings()
+        assert s.APP_NAME == s.app_name
+
+    def test_app_version_alias(self):
+        s = APISettings()
+        assert s.APP_VERSION == s.app_version
+
+    def test_database_url_alias(self):
+        s = APISettings()
+        assert s.DATABASE_URL == s.database_url
+
+    def test_redis_url_alias(self):
+        s = APISettings()
+        assert s.REDIS_URL == s.redis_url
+
+    def test_jwt_secret_alias(self):
+        s = APISettings()
+        assert s.JWT_SECRET_KEY == s.jwt_secret_key
+
+    def test_jwt_algorithm_alias(self):
+        s = APISettings()
+        assert s.JWT_ALGORITHM == s.jwt_algorithm
+
+    def test_smtp_host_alias(self):
+        s = APISettings()
+        assert s.SMTP_HOST == s.smtp_host
+
+    def test_oss_endpoint_alias(self):
+        s = APISettings()
+        assert s.OSS_ENDPOINT == s.oss_endpoint
+
+    def test_celery_broker_alias(self):
+        s = APISettings()
+        assert s.CELERY_BROKER_URL == s.celery_broker_url
+
+    def test_render_engine_alias(self):
+        s = APISettings()
+        assert s.RENDER_ENGINE == s.render_engine
+
+
+class TestSettingsCache:
+    """配置单例缓存测试."""
+
+    def test_get_api_settings_returns_same_instance(self):
+        s1 = get_api_settings()
+        s2 = get_api_settings()
+        assert s1 is s2
+
+    def test_get_cached_settings_same_class_same_instance(self):
+        s1 = get_cached_settings(SharedSettings)
+        s2 = get_cached_settings(SharedSettings)
+        assert s1 is s2
+
+    def test_reload_clears_cache(self):
+        s1 = get_cached_settings(SharedSettings)
+        reload_settings_cache()
+        s2 = get_cached_settings(SharedSettings)
+        assert s1 is not s2
+
+    def test_custom_cache_key(self):
+        s1 = get_cached_settings(SharedSettings, cache_key="custom1")
+        s2 = get_cached_settings(SharedSettings, cache_key="custom2")
+        assert s1 is not s2
+
+    def test_get_shared_settings(self):
+        from packages.config.base import get_shared_settings
+        s = get_shared_settings()
+        assert isinstance(s, SharedSettings)
+
+
+class TestOSSAliases:
+    """OSS 配置别名测试."""
+
+    def test_oss_bucket_name_alias(self):
+        s = APISettings()
+        assert s.OSS_BUCKET_NAME == s.oss_bucket_name
+
+    def test_oss_direct_upload_max_mb_alias(self):
+        s = APISettings()
+        assert s.OSS_DIRECT_UPLOAD_MAX_MB == s.oss_direct_upload_max_mb
+
+    def test_oss_direct_upload_expire_alias(self):
+        s = APISettings()
+        assert s.OSS_DIRECT_UPLOAD_EXPIRE_SECONDS == s.oss_direct_upload_expire_seconds
+
+# ── WorkerSettings 测试 ──────────────────────────────────────
+
+from packages.config.worker_settings import WorkerSettings, get_worker_settings
+
+
+class TestWorkerSettingsDefaults:
+    """WorkerSettings 默认值测试."""
+
+    def test_default_worker_name(self):
+        s = WorkerSettings()
+        assert s.worker_name == "xiaoxia-saas-worker"
+
+    def test_default_worker_concurrency(self):
+        s = WorkerSettings()
+        assert s.worker_concurrency == 4
+
+    def test_default_worker_max_tasks_per_child(self):
+        s = WorkerSettings()
+        assert s.worker_max_tasks_per_child == 1000
+
+    def test_broker_url_alias(self):
+        s = WorkerSettings()
+        assert s.broker_url == s.celery_broker_url
+
+    def test_result_backend_alias(self):
+        s = WorkerSettings()
+        assert s.result_backend == s.celery_result_backend
+
+    def test_inherits_shared_settings(self):
+        s = WorkerSettings()
+        assert s.database_url  # 继承自SharedSettings
+        assert s.redis_url
+        assert s.oss_endpoint
+        assert s.cosyvoice_model == "cosyvoice-v3-flash"
+
+
+class TestGetWorkerSettings:
+    """get_worker_settings 单例测试."""
+
+    def test_returns_worker_settings_instance(self):
+        s = get_worker_settings()
+        assert isinstance(s, WorkerSettings)
+
+    def test_singleton(self):
+        s1 = get_worker_settings()
+        s2 = get_worker_settings()
+        assert s1 is s2
diff --git a/tests/unit/test_tts_streaming.py b/tests/unit/test_tts_streaming.py
index 1f587aedd..62413d07b 100755
--- a/tests/unit/test_tts_streaming.py
+++ b/tests/unit/test_tts_streaming.py
@@ -232,3 +232,215 @@ class TestTTSStreamingService:
 
             assert result == b"audio data"
             mock_download.assert_called_once()
+
+
+class TestTTSStreamingEdgeCases:
+    """流式合成边界测试."""
+
+    @pytest.mark.asyncio
+    async def test_exactly_500_chars_uses_short_text(self):
+        """刚好500字走短文本路径."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 5.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "x" * 500, "voice_id": "test_voice"}
+
+        with patch.object(service, "_download_audio", return_value=b"audio"):
+            await service.synthesize_and_stream(ws, params)
+
+        # 短文本只有1个segment
+        assert ws.sent_json[0]["type"] == "started"
+        assert ws.sent_json[0]["segment_count"] == 1
+
+    @pytest.mark.asyncio
+    async def test_501_chars_uses_long_text(self):
+        """501字走长文本分段路径."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 2.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "x" * 501, "voice_id": "test_voice"}
+
+        with patch.object(service, "_download_audio", return_value=b"audio"):
+            await service.synthesize_and_stream(ws, params)
+
+        # 长文本segment_count > 1
+        assert ws.sent_json[0]["type"] == "started"
+        assert ws.sent_json[0]["segment_count"] >= 2
+
+    @pytest.mark.asyncio
+    async def test_short_text_speed_param_passed(self):
+        """短文本合成时速度参数正确传递."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 3.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "测试", "voice_id": "v1", "speed": 1.5, "format": "wav"}
+
+        with patch.object(service, "_download_audio", return_value=b"audio"):
+            await service.synthesize_and_stream(ws, params)
+
+        cosyvoice.submit_synthesize_task.assert_called_once()
+        call_kwargs = cosyvoice.submit_synthesize_task.call_args.kwargs
+        assert call_kwargs["speed"] == 1.5
+        assert call_kwargs["format"] == "wav"
+        assert call_kwargs["voice_id"] == "v1"
+
+    @pytest.mark.asyncio
+    async def test_short_text_sample_rate_param(self):
+        """短文本合成时采样率参数传递."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 1.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "测试", "voice_id": "v1", "sample_rate": 44100}
+
+        with patch.object(service, "_download_audio", return_value=b"audio"):
+            await service.synthesize_and_stream(ws, params)
+
+        call_kwargs = cosyvoice.submit_synthesize_task.call_args.kwargs
+        assert call_kwargs["sample_rate"] == 44100
+
+    @pytest.mark.asyncio
+    async def test_single_chunk_audio(self):
+        """小于4KB的音频只发1块."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        audio_data = b"x" * 1000  # 1KB < 4KB
+
+        total = await service._stream_audio_chunks(ws, audio_data)
+        assert total == 1000
+        assert len(ws.sent_bytes) == 1
+        assert ws.sent_bytes[0] == audio_data
+
+    @pytest.mark.asyncio
+    async def test_exact_chunk_size_audio(self):
+        """刚好4KB的音频只发1块."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        audio_data = b"x" * 4096
+
+        total = await service._stream_audio_chunks(ws, audio_data)
+        assert total == 4096
+        assert len(ws.sent_bytes) == 1
+
+    @pytest.mark.asyncio
+    async def test_empty_audio_chunks(self):
+        """空音频数据不发送任何块."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+
+        total = await service._stream_audio_chunks(ws, b"")
+        assert total == 0
+        assert len(ws.sent_bytes) == 0
+
+    @pytest.mark.asyncio
+    async def test_short_text_unexpected_exception(self):
+        """短文本合成时非预期异常捕获."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.side_effect = RuntimeError("Unexpected error")
+
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "测试", "voice_id": "v1"}
+
+        await service.synthesize_and_stream(ws, params)
+
+        assert ws.sent_json[-1]["type"] == "error"
+        assert "合成失败" in ws.sent_json[-1]["message"]
+
+    @pytest.mark.asyncio
+    async def test_short_text_download_failure(self):
+        """短文本音频下载失败."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/audio.mp3",
+            "duration": 1.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "测试", "voice_id": "v1"}
+
+        with patch.object(service, "_download_audio", side_effect=Exception("Download failed")):
+            await service.synthesize_and_stream(ws, params)
+
+        assert ws.sent_json[-1]["type"] == "error"
+        assert "音频推送失败" in ws.sent_json[-1]["message"]
+
+    @pytest.mark.asyncio
+    async def test_long_text_segment_count_matches_split(self):
+        """长文本分段数量与split_text结果一致."""
+        from packages.application.tts_job.text_splitter import split_text
+
+        text = "x" * 1200
+        segments = split_text(text, max_chars=500)
+        expected_count = len(segments)
+
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/a.mp3",
+            "duration": 1.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": text, "voice_id": "v1"}
+
+        with patch.object(service, "_download_audio", return_value=b"audio"):
+            await service.synthesize_and_stream(ws, params)
+
+        assert ws.sent_json[0]["segment_count"] == expected_count
+        segment_done = sum(1 for m in ws.sent_json if m["type"] == "segment_done")
+        assert segment_done == expected_count
+
+    @pytest.mark.asyncio
+    async def test_long_text_total_bytes_accumulated(self):
+        """长文本总字节数正确累加."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        cosyvoice.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/a.mp3",
+            "duration": 1.0,
+        }
+        service = TTSStreamingService(cosyvoice)
+        ws = MockWebSocket()
+        params = {"text": "x" * 600, "voice_id": "v1"}
+
+        audio_chunk = b"x" * 5000
+        with patch.object(service, "_download_audio", return_value=audio_chunk):
+            await service.synthesize_and_stream(ws, params)
+
+        # done帧中file_size应为分段数 * 每段大小
+        done_msg = ws.sent_json[-1]
+        assert done_msg["type"] == "done"
+        segment_count = ws.sent_json[0]["segment_count"]
+        assert done_msg["file_size"] == segment_count * 5000
+
+    def test_download_audio_passes_purpose_and_mime(self):
+        """_download_audio正确传递参数给safe_download_bytes."""
+        cosyvoice = MagicMock(spec=CosyVoiceService)
+        service = TTSStreamingService(cosyvoice)
+
+        with patch("packages.application.tts_job.streaming_service.safe_download_bytes") as mock:
+            mock.return_value = b"data"
+            service._download_audio("https://example.com/a.wav")
+
+            mock.assert_called_once()
+            kwargs = mock.call_args.kwargs
+            assert kwargs["purpose"] == "tts_streaming_download"
+            assert kwargs["timeout"] == 60.0
+            assert "allowed_mime_types" in kwargs
-- 
2.54.0