xiaoxia
37d111dcd5
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
84 lines
2.8 KiB
Python
Executable File
84 lines
2.8 KiB
Python
Executable File
"""
|
|
Worker 配置测试.
|
|
|
|
覆盖 WorkerSettings 默认值、属性别名等纯逻辑.
|
|
环境变量加载由集成测试覆盖.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.config.worker_settings import WorkerSettings
|
|
|
|
|
|
class TestWorkerSettingsDefaults:
|
|
"""WorkerSettings 默认值."""
|
|
|
|
def test_worker_name_default(self):
|
|
settings = WorkerSettings()
|
|
assert settings.worker_name == "xiaoxia-saas-worker"
|
|
|
|
def test_worker_concurrency_default(self):
|
|
settings = WorkerSettings()
|
|
assert settings.worker_concurrency == 4
|
|
|
|
def test_worker_max_tasks_per_child_default(self):
|
|
settings = WorkerSettings()
|
|
assert settings.worker_max_tasks_per_child == 1000
|
|
|
|
def test_inherits_shared_settings(self):
|
|
"""继承 SharedSettings 的字段."""
|
|
settings = WorkerSettings()
|
|
# 验证至少有一些 SharedSettings 的字段存在
|
|
assert hasattr(settings, "celery_broker_url")
|
|
assert hasattr(settings, "celery_result_backend")
|
|
|
|
def test_custom_values(self):
|
|
settings = WorkerSettings(
|
|
worker_name="test-worker",
|
|
worker_concurrency=8,
|
|
worker_max_tasks_per_child=500,
|
|
)
|
|
assert settings.worker_name == "test-worker"
|
|
assert settings.worker_concurrency == 8
|
|
assert settings.worker_max_tasks_per_child == 500
|
|
|
|
|
|
class TestWorkerSettingsAliases:
|
|
"""Celery 字段名向后兼容别名."""
|
|
|
|
def test_broker_url_alias(self):
|
|
settings = WorkerSettings(celery_broker_url="redis://localhost:6379/0")
|
|
assert settings.broker_url == settings.celery_broker_url
|
|
assert settings.broker_url == "redis://localhost:6379/0"
|
|
|
|
def test_result_backend_alias(self):
|
|
settings = WorkerSettings(celery_result_backend="redis://localhost:6379/1")
|
|
assert settings.result_backend == settings.celery_result_backend
|
|
assert settings.result_backend == "redis://localhost:6379/1"
|
|
|
|
def test_broker_url_is_property(self):
|
|
"""broker_url 是 property，每次读取都返回最新值."""
|
|
settings = WorkerSettings()
|
|
# 验证是 property 描述符
|
|
assert isinstance(type(settings).broker_url, property)
|
|
assert isinstance(type(settings).result_backend, property)
|
|
|
|
|
|
class TestWorkerSettingsType:
|
|
"""类型验证."""
|
|
|
|
def test_worker_concurrency_is_int(self):
|
|
settings = WorkerSettings()
|
|
assert isinstance(settings.worker_concurrency, int)
|
|
|
|
def test_worker_max_tasks_is_int(self):
|
|
settings = WorkerSettings()
|
|
assert isinstance(settings.worker_max_tasks_per_child, int)
|
|
|
|
def test_worker_name_is_str(self):
|
|
settings = WorkerSettings()
|
|
assert isinstance(settings.worker_name, str)
|
|
assert len(settings.worker_name) > 0
|