144 lines
4.8 KiB
Python
Executable File
144 lines
4.8 KiB
Python
Executable File
"""mark_title_used_for_generation 纯逻辑单测.
|
|
|
|
验证 title usage 计数 + updated_at 更新逻辑。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock
|
|
|
|
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()
|