Xiaoxia Agent
d0ff504a5f
核心改动： - packages/shared/ai_config_version.py: Redis版本号通知机制 - packages/shared/ai_router.py: AIRouter统一路由层（DB→Redis缓存→SharedSettings fallback） - alembic/versions/099: 补齐缺失模型seed和capability配置 - apps/worker/worker_app/tasks/vision/*: VLM硬编码替换为ai_router动态配置 - packages/application/viral_video/reviewer.py: 审核走copy_review capability - packages/application/cosyvoice_service.py: TTS走tts capability - apps/worker/worker_app/tasks/viral_video.py: 意图解析/分镜/文案走DB配置 - packages/config/base.py: 移除doubao/dashscope/mediakit/cosyvoice硬编码默认值 - tests/unit/test_ai_router.py: 26个单元测试覆盖路由/缓存/fallback Admin侧： - app/utils/ai_config_notify.py: bump_ai_config_version()工具函数 - routers/ai_models.py + ai_capability_configs.py: CRUD后调用bump_version() 验收标准： - admin改模型后30s内生效（Redis版本号通知） - Redis故障fallback环境变量 - base.py零硬编码（api_key保留空串） - 26个单测全绿
338 lines
11 KiB
Python
Executable File
338 lines
11 KiB
Python
Executable File
"""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 == "" # 零硬编码：默认值已清空
|
|
assert s.cosyvoice_format == "mp3"
|
|
assert s.cosyvoice_sample_rate == 22050
|
|
|
|
def test_default_doubao_settings(self):
|
|
s = SharedSettings()
|
|
assert s.doubao_model == "" # 零硬编码：默认值已清空
|
|
assert s.doubao_timeout == 45 # #2180 默认提到45s
|
|
assert s.doubao_max_retries == 1
|
|
|
|
|
|
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 == "" # 零硬编码：默认值已清空
|
|
|
|
|
|
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
|