xiaoxia-saas/tests/unit/test_config_base.py

"""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