"""Config Base 单元测试""" from __future__ import annotations from unittest.mock import MagicMock import pytest from packages.config.base import ( SharedSettings, get_cached_settings, get_shared_settings, reload_settings_cache, ) class TestSharedSettingsDefaults: """SharedSettings 默认值测试""" @pytest.fixture(autouse=True) def clean_env(self, monkeypatch): """清除所有可能影响的环境变量，确保测的是代码默认值""" env_vars = [ "ENVIRONMENT", "DEBUG", "AUTO_CREATE_SCHEMA", "DATABASE_URL", "DATABASE_POOL_SIZE", "DATABASE_MAX_OVERFLOW", "DATABASE_POOL_TIMEOUT", "DATABASE_POOL_RECYCLE", "REDIS_URL", "CELERY_BROKER_URL", "CELERY_RESULT_BACKEND", "OSS_ENDPOINT", "OSS_ACCESS_KEY_ID", "OSS_ACCESS_KEY_SECRET", "OSS_BUCKET_NAME", "OSS_DIRECT_UPLOAD_MAX_MB", "OSS_DIRECT_UPLOAD_EXPIRE_SECONDS", "COSYVOICE_API_KEY", "COSYVOICE_BASE_URL", "COSYVOICE_MODEL", "COSYVOICE_VOICE", "COSYVOICE_SAMPLE_RATE", "COSYVOICE_FORMAT", "COSYVOICE_CLONE_MODEL", "DOUBAO_API_KEY", "DOUBAO_MODEL", "DOUBAO_BASE_URL", "DOUBAO_TIMEOUT", "DOUBAO_MAX_RETRIES", ] for var in env_vars: monkeypatch.delenv(var, raising=False) reload_settings_cache() yield reload_settings_cache() def _make_settings(self): """构造不读 env 文件的纯净 settings""" return SharedSettings(_env_file="/dev/null") def test_default_environment(self): """默认环境为 development""" s = self._make_settings() assert s.environment == "development" def test_default_debug(self): """默认开启 debug""" s = self._make_settings() assert s.debug is True def test_default_database_config(self): """数据库默认配置""" s = self._make_settings() assert "postgresql" in s.database_url assert s.database_pool_size == 20 assert s.database_max_overflow == 10 assert s.database_pool_timeout == 30 assert s.database_pool_recycle == 3600 def test_default_redis_config(self): """Redis 默认配置""" s = self._make_settings() assert s.redis_url.startswith("redis://") def test_default_celery_config(self): """Celery 默认配置""" s = self._make_settings() assert s.celery_broker_url.startswith("redis://") assert s.celery_result_backend.startswith("redis://") def test_default_oss_config(self): """OSS 默认配置""" s = self._make_settings() assert s.oss_endpoint.endswith("aliyuncs.com") assert s.oss_bucket_name == "xiaoxia-autocut" assert s.oss_direct_upload_max_mb == 2000 assert s.oss_direct_upload_expire_seconds == 900 def test_default_cosyvoice_config(self): """CosyVoice 默认配置""" s = self._make_settings() assert s.cosyvoice_model == "cosyvoice-v3-flash" assert s.cosyvoice_sample_rate == 22050 assert s.cosyvoice_format == "mp3" assert s.cosyvoice_clone_model == "voice-enrollment" def test_default_doubao_config(self): """豆包默认配置""" s = self._make_settings() assert s.doubao_timeout == 30 assert s.doubao_max_retries == 2 assert "volces.com" in s.doubao_base_url def test_default_empty_api_keys(self): """API Key 默认空字符串""" s = self._make_settings() assert s.oss_access_key_id == "" assert s.oss_access_key_secret == "" assert s.cosyvoice_api_key == "" assert s.doubao_api_key == "" def test_auto_create_schema_default(self): """auto_create_schema 默认 False""" s = self._make_settings() assert s.auto_create_schema is False class TestSettingsSingleton: """单例管理测试""" def setup_method(self): """每个测试前清空缓存""" reload_settings_cache() def teardown_method(self): """每个测试后清空缓存""" reload_settings_cache() def test_get_cached_settings_same_instance(self): """同一类两次调用返回同一实例""" s1 = get_cached_settings(SharedSettings) s2 = get_cached_settings(SharedSettings) assert s1 is s2 def test_get_shared_settings_returns_shared_settings(self): """get_shared_settings 返回 SharedSettings 实例""" s = get_shared_settings() assert isinstance(s, SharedSettings) def test_get_shared_settings_singleton(self): """get_shared_settings 是单例""" s1 = get_shared_settings() s2 = get_shared_settings() assert s1 is s2 def test_reload_settings_cache_clears(self): """reload 后获取新实例""" s1 = get_cached_settings(SharedSettings) reload_settings_cache() s2 = get_cached_settings(SharedSettings) assert s1 is not s2 def test_custom_cache_key(self): """自定义 cache_key 分开缓存""" s1 = get_cached_settings(SharedSettings, cache_key="key_a") s2 = get_cached_settings(SharedSettings, cache_key="key_b") assert s1 is not s2 # 但值相同 assert s1.database_url == s2.database_url def test_different_classes_separate_cache(self): """不同类使用不同缓存""" from packages.config.api_settings import APISettings shared = get_shared_settings() api = get_cached_settings(APISettings) assert shared is not api class TestPointsEnabledSwitch: """#1895 P2: POINTS_ENABLED 配置开关（默认 false 保护现有用户）。""" def test_default_points_enabled_is_false(self): from packages.config.base import SharedSettings s = SharedSettings() assert s.points_enabled is False def test_points_enabled_can_be_set_true(self, monkeypatch): from packages.config import base as base_mod monkeypatch.setenv("POINTS_ENABLED", "true") base_mod.reload_settings_cache() try: s = base_mod.SharedSettings() assert s.points_enabled is True finally: monkeypatch.delenv("POINTS_ENABLED", raising=False) base_mod.reload_settings_cache() def test_points_gate_disabled_passthrough(self, monkeypatch): """开关关闭时，@points_gate 装饰器完全透传原函数。""" import packages.middleware.points_gate as pg_mod from packages.middleware.points_gate import points_gate monkeypatch.setattr(pg_mod, "_points_gate_enabled", lambda: False) @points_gate("ai_rewrite") def my_func(current_user=None, db=None): return "bypass" # 不传 current_user/db 也不报错（证明扣点逻辑被跳过） assert my_func() == "bypass" def test_points_gate_enabled_blocks_without_user(self, monkeypatch): """开关开启时，没有 current_user 会抛 401。""" from fastapi import HTTPException import packages.middleware.points_gate as pg_mod from packages.middleware.points_gate import points_gate monkeypatch.setattr(pg_mod, "_points_gate_enabled", lambda: True) @points_gate("ai_rewrite") def my_func(current_user=None, db=None): return "ok" with pytest.raises(HTTPException) as exc: my_func(db=MagicMock()) assert exc.value.status_code == 401