xiaoxia-saas/tests/unit/test_title_usage_pure.py

"""mark_title_used_for_generation 纯逻辑单测.

验证 title usage 计数 + updated_at 更新逻辑。
"""

from __future__ import annotations

from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

import pytest

from packages.adapters.sqlalchemy_impl.models import TitleLibraryModel


def test_module_importable():
    """确认模块可以正常导入."""
    from worker_app.core.title_usage import mark_title_used_for_generation  # noqa: F401


class TestMarkTitleUsedForGeneration:
    """mark_title_used_for_generation 测试."""

    def _make_task(self, strategy_id: str = "title-1") -> MagicMock:
        task = MagicMock()
        task.strategy_id = strategy_id
        return task

    def _make_title(self, usage_count: int = 0) -> MagicMock:
        title = MagicMock(spec=TitleLibraryModel)
        title.id = "title-1"
        title.usage_count = usage_count
        title.updated_at = None
        return title

    def test_no_strategy_id_returns_early(self):
        """无 strategy_id 时直接返回，不查 DB."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        task = self._make_task(strategy_id="")

        mark_title_used_for_generation(db, task)

        db.query.assert_not_called()

    def test_none_strategy_id_returns_early(self):
        """strategy_id 为 None 时直接返回."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        task = self._make_task(strategy_id=None)

        mark_title_used_for_generation(db, task)

        db.query.assert_not_called()

    def test_title_not_found_returns_early(self):
        """title 不存在时不报错，静默返回."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = None
        task = self._make_task(strategy_id="missing-id")

        mark_title_used_for_generation(db, task)

        db.add.assert_not_called()
        db.commit.assert_not_called()

    def test_increments_usage_count_from_zero(self):
        """usage_count 从 0 递增到 1."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        title = self._make_title(usage_count=0)
        db.query.return_value.filter.return_value.first.return_value = title
        task = self._make_task()

        mark_title_used_for_generation(db, task)

        assert title.usage_count == 1
        db.add.assert_called_once_with(title)
        db.commit.assert_called_once()

    def test_increments_usage_count_from_existing(self):
        """已有 usage_count 时递增."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        title = self._make_title(usage_count=5)
        db.query.return_value.filter.return_value.first.return_value = title
        task = self._make_task()

        mark_title_used_for_generation(db, task)

        assert title.usage_count == 6

    def test_none_usage_count_defaults_to_zero_then_increments(self):
        """usage_count 为 None 时按 0 处理，递增到 1."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        title = self._make_title(usage_count=None)
        db.query.return_value.filter.return_value.first.return_value = title
        task = self._make_task()

        mark_title_used_for_generation(db, task)

        assert title.usage_count == 1

    def test_updates_updated_at_to_utc_now(self):
        """updated_at 更新为当前 UTC 时间."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        title = self._make_title(usage_count=3)
        db.query.return_value.filter.return_value.first.return_value = title
        task = self._make_task()

        before = datetime.now(timezone.utc)
        mark_title_used_for_generation(db, task)
        after = datetime.now(timezone.utc)

        assert before <= title.updated_at <= after
        assert title.updated_at.tzinfo is not None  # 带时区

    def test_correct_query_filter(self):
        """查询时使用正确的 id 过滤."""
        from worker_app.core.title_usage import mark_title_used_for_generation

        db = MagicMock()
        title = self._make_title()
        db.query.return_value.filter.return_value.first.return_value = title
        task = self._make_task(strategy_id="title-abc")

        mark_title_used_for_generation(db, task)

        # 验证 query 模型正确
        db.query.assert_called_once_with(TitleLibraryModel)
        # 验证 filter 条件
        filter_call = db.query.return_value.filter
        assert filter_call.called
        # first 被调用
        filter_call.return_value.first.assert_called_once()