xiaoxia-saas/tests/unit/test_1286_preview_count_validation.py

"""预览数量 preview_count 校验测试。

验证 CreatePreviewGenerationTaskRequest 中 preview_count 字段：
- 默认值为 1
- 范围 1-10
- 超出范围报错
- Worker 并发配置验证
"""

import pytest


class TestPreviewCountValidation:
    """preview_count 字段校验"""

    def test_default_preview_count_is_1(self):
        """默认 preview_count 为 1"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["asset_1"],
        )
        assert req.preview_count == 1

    def test_preview_count_min_valid(self):
        """preview_count=1 合法"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["asset_1"],
            preview_count=1,
        )
        assert req.preview_count == 1

    def test_preview_count_max_valid(self):
        """preview_count=10 合法（上限）"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["asset_1"],
            preview_count=10,
        )
        assert req.preview_count == 10

    def test_preview_count_middle_value(self):
        """preview_count=5 合法"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["asset_1"],
            preview_count=5,
        )
        assert req.preview_count == 5

    def test_preview_count_zero_raises(self):
        """preview_count=0 报错（低于下限）"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            CreatePreviewGenerationTaskRequest(
                template_id="tmpl_123",
                asset_ids=["asset_1"],
                preview_count=0,
            )

    def test_preview_count_negative_raises(self):
        """preview_count=-1 报错"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            CreatePreviewGenerationTaskRequest(
                template_id="tmpl_123",
                asset_ids=["asset_1"],
                preview_count=-1,
            )

    def test_preview_count_exceeds_max_raises(self):
        """preview_count=11 报错（超过上限 10）"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            CreatePreviewGenerationTaskRequest(
                template_id="tmpl_123",
                asset_ids=["asset_1"],
                preview_count=11,
            )

    def test_preview_count_large_value_raises(self):
        """preview_count=100 报错"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
        from pydantic import ValidationError

        with pytest.raises(ValidationError):
            CreatePreviewGenerationTaskRequest(
                template_id="tmpl_123",
                asset_ids=["asset_1"],
                preview_count=100,
            )


class TestWorkerConcurrencyConfig:
    """Worker 并发配置验证"""

    def test_default_worker_concurrency_is_4(self):
        """Worker 默认并发为 4"""
        from packages.config import WorkerSettings

        settings = WorkerSettings()
        assert settings.worker_concurrency == 4

    def test_worker_concurrency_configurable(self):
        """Worker 并发可通过环境变量配置"""
        from packages.config import WorkerSettings

        settings = WorkerSettings(worker_concurrency=8)
        assert settings.worker_concurrency == 8

    def test_worker_concurrency_is_int(self):
        """Worker 并发为整数类型"""
        from packages.config import WorkerSettings

        settings = WorkerSettings()
        assert isinstance(settings.worker_concurrency, int)