xiaoxia-saas/tests/unit/test_worker_settings.py

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