xiaoxia-saas/tests/unit/test_api_settings.py

"""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()
    # 保存关键环境变量（避免其他测试模块的全局污染）
    _saved_env = {}
    for key in ["JWT_SECRET_KEY", "DATABASE_URL", "USE_IN_MEMORY_DB", "APP_ENV"]:
        _saved_env[key] = os.environ.get(key)
    # 设置必要的环境变量，避免 JWT 校验失败
    os.environ["JWT_SECRET_KEY"] = "test-secret-key-for-unit-tests-only-12345"
    # 清除可能被其他模块污染的变量，确保默认值测试准确
    for key in ["DATABASE_URL", "APP_ENV"]:
        os.environ.pop(key, None)
    yield
    reload_settings_cache()
    # 恢复所有保存的环境变量，避免污染其他测试模块
    for key, val in _saved_env.items():
        if val is None:
            os.environ.pop(key, None)
        else:
            os.environ[key] = val


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