140 lines
5.1 KiB
Python
Executable File
140 lines
5.1 KiB
Python
Executable File
"""mark_asset_used_for_generation 深度补充单测.
|
|
|
|
补全边界场景：空 metadata、None metadata、last_used_at 格式、
|
|
review_status 已有值不覆盖、多次调用递增。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from worker_app.core.asset_usage import mark_asset_used_for_generation
|
|
|
|
from packages.domain import Asset, AssetStatus
|
|
|
|
|
|
def _asset() -> Asset:
|
|
return Asset.create(
|
|
project_id="project-1",
|
|
library_id="library-1",
|
|
name="video.mp4",
|
|
storage_key="uploads/video.mp4",
|
|
mime_type="video/mp4",
|
|
file_size=1024,
|
|
status=AssetStatus.READY,
|
|
)
|
|
|
|
|
|
class TestMarkAssetUsedForGeneration:
|
|
"""mark_asset_used_for_generation 深度测试."""
|
|
|
|
def test_first_use_sets_count_to_1(self):
|
|
"""首次使用，use_count 从 0 变 1."""
|
|
asset = _asset()
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 1
|
|
|
|
def test_increments_existing_count(self):
|
|
"""已有计数时递增."""
|
|
asset = _asset()
|
|
asset.metadata = {"generation_use_count": 5}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 6
|
|
|
|
def test_zero_count_increments_to_1(self):
|
|
"""计数为 0 时递增到 1."""
|
|
asset = _asset()
|
|
asset.metadata = {"generation_use_count": 0}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 1
|
|
|
|
def test_empty_metadata_still_works(self):
|
|
"""空 dict metadata 也能正常工作."""
|
|
asset = _asset()
|
|
asset.metadata = {}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 1
|
|
assert asset.metadata["review_status"] == "pending_review"
|
|
assert "last_used_at" in asset.metadata
|
|
|
|
def test_none_metadata_field_defaults_to_0(self):
|
|
"""metadata 中 generation_use_count 为 None 时按 0 处理."""
|
|
asset = _asset()
|
|
asset.metadata = {"generation_use_count": None}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 1
|
|
|
|
def test_string_count_gets_casted(self):
|
|
"""字符串类型的 use_count 通过 int() 转换."""
|
|
asset = _asset()
|
|
asset.metadata = {"generation_use_count": "3"}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 4
|
|
|
|
def test_preserves_other_metadata_fields(self):
|
|
"""不覆盖 metadata 中的其他字段."""
|
|
asset = _asset()
|
|
asset.metadata = {
|
|
"generation_use_count": 1,
|
|
"custom_field": "value",
|
|
"tags": ["a", "b"],
|
|
}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["custom_field"] == "value"
|
|
assert asset.metadata["tags"] == ["a", "b"]
|
|
assert asset.metadata["generation_use_count"] == 2
|
|
|
|
def test_review_status_pending_when_not_set(self):
|
|
"""review_status 未设置时设为 pending_review."""
|
|
asset = _asset()
|
|
asset.metadata = {}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["review_status"] == "pending_review"
|
|
|
|
def test_review_status_not_overwritten_if_present(self):
|
|
"""review_status 已有值时不覆盖."""
|
|
asset = _asset()
|
|
asset.metadata = {"review_status": "approved"}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["review_status"] == "approved"
|
|
|
|
def test_review_status_empty_string_considered_falsy(self):
|
|
"""review_status 为空字符串时视为 falsy，设置为 pending_review."""
|
|
asset = _asset()
|
|
asset.metadata = {"review_status": ""}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["review_status"] == "pending_review"
|
|
|
|
def test_last_used_at_is_iso_format(self):
|
|
"""last_used_at 是 ISO 格式时间字符串."""
|
|
asset = _asset()
|
|
mark_asset_used_for_generation(asset)
|
|
ts = asset.metadata["last_used_at"]
|
|
# 可以被解析为 ISO 格式
|
|
parsed = datetime.fromisoformat(ts)
|
|
assert parsed.tzinfo is not None # 带时区
|
|
|
|
def test_last_used_at_is_utc(self):
|
|
"""last_used_at 是 UTC 时间."""
|
|
asset = _asset()
|
|
before = datetime.now(timezone.utc)
|
|
mark_asset_used_for_generation(asset)
|
|
after = datetime.now(timezone.utc)
|
|
ts = datetime.fromisoformat(asset.metadata["last_used_at"])
|
|
assert before <= ts <= after
|
|
|
|
def test_multiple_calls_increment_count(self):
|
|
"""多次调用持续递增."""
|
|
asset = _asset()
|
|
mark_asset_used_for_generation(asset)
|
|
mark_asset_used_for_generation(asset)
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == 3
|
|
|
|
def test_negative_count_still_increments(self):
|
|
"""负数计数（异常数据）也能递增."""
|
|
asset = _asset()
|
|
asset.metadata = {"generation_use_count": -5}
|
|
mark_asset_used_for_generation(asset)
|
|
assert asset.metadata["generation_use_count"] == -4
|