Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bc324baf3e | |||
| 3d274eaa87 |
Executable
+290
@@ -0,0 +1,290 @@
|
||||
"""media_validation 媒体文件校验单测."""
|
||||
|
||||
import pytest
|
||||
from domain.media_validation import (
|
||||
MIN_AUDIO_FILE_SIZE,
|
||||
MIN_IMAGE_FILE_SIZE,
|
||||
MIN_VIDEO_FILE_SIZE,
|
||||
SUPPORTED_VIDEO_CODECS,
|
||||
is_valid_media,
|
||||
safe_parse_fps,
|
||||
)
|
||||
|
||||
# ── 常量测试 ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestConstants:
|
||||
"""模块常量"""
|
||||
|
||||
def test_min_sizes(self):
|
||||
assert MIN_VIDEO_FILE_SIZE == 1024
|
||||
assert MIN_AUDIO_FILE_SIZE == 100
|
||||
assert MIN_IMAGE_FILE_SIZE == 100
|
||||
|
||||
def test_supported_codecs_is_frozenset(self):
|
||||
assert isinstance(SUPPORTED_VIDEO_CODECS, frozenset)
|
||||
|
||||
def test_supported_codecs_includes_common(self):
|
||||
assert "h264" in SUPPORTED_VIDEO_CODECS
|
||||
assert "hevc" in SUPPORTED_VIDEO_CODECS
|
||||
assert "vp9" in SUPPORTED_VIDEO_CODECS
|
||||
assert "av1" in SUPPORTED_VIDEO_CODECS
|
||||
assert "mpeg4" in SUPPORTED_VIDEO_CODECS
|
||||
assert "prores" in SUPPORTED_VIDEO_CODECS
|
||||
|
||||
def test_supported_codecs_count(self):
|
||||
assert len(SUPPORTED_VIDEO_CODECS) >= 20
|
||||
|
||||
|
||||
# ── safe_parse_fps ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSafeParseFps:
|
||||
"""safe_parse_fps 函数"""
|
||||
|
||||
def test_simple_decimal(self):
|
||||
assert safe_parse_fps("30.0") == 30.0
|
||||
|
||||
def test_integer_string(self):
|
||||
assert safe_parse_fps("24") == 24.0
|
||||
|
||||
def test_fraction_format(self):
|
||||
assert abs(safe_parse_fps("30000/1001") - 29.97) < 0.01
|
||||
|
||||
def test_simple_fraction(self):
|
||||
assert safe_parse_fps("30/1") == 30.0
|
||||
|
||||
def test_24fps_fraction(self):
|
||||
assert safe_parse_fps("24/1") == 24.0
|
||||
|
||||
def test_60fps_fraction(self):
|
||||
assert safe_parse_fps("60000/1001") == pytest.approx(59.94, abs=0.01)
|
||||
|
||||
def test_zero_denominator_returns_zero(self):
|
||||
assert safe_parse_fps("30/0") == 0.0
|
||||
|
||||
def test_empty_string_returns_zero(self):
|
||||
assert safe_parse_fps("") == 0.0
|
||||
|
||||
def test_invalid_string_returns_zero(self):
|
||||
assert safe_parse_fps("invalid") == 0.0
|
||||
|
||||
def test_none_numerator_fraction(self):
|
||||
assert safe_parse_fps("abc/1001") == 0.0
|
||||
|
||||
def test_negative_fps(self):
|
||||
assert safe_parse_fps("-30") == -30.0
|
||||
|
||||
def test_very_high_fps(self):
|
||||
assert safe_parse_fps("240/1") == 240.0
|
||||
|
||||
def test_multiple_slashes(self):
|
||||
# 只按第一个 / 分割
|
||||
# "30/1/2" → num="30", den="1/2" → float("1/2") 抛异常 → 返回 0
|
||||
assert safe_parse_fps("30/1/2") == 0.0
|
||||
|
||||
def test_float_fraction(self):
|
||||
result = safe_parse_fps("29.97/1")
|
||||
assert result == pytest.approx(29.97)
|
||||
|
||||
def test_zero_fps(self):
|
||||
assert safe_parse_fps("0") == 0.0
|
||||
|
||||
def test_zero_numerator(self):
|
||||
assert safe_parse_fps("0/1000") == 0.0
|
||||
|
||||
|
||||
# ── is_valid_media - video ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsValidMediaVideo:
|
||||
"""is_valid_media 视频校验"""
|
||||
|
||||
def test_valid_video(self):
|
||||
metadata = {
|
||||
"size_bytes": 1024 * 1024, # 1MB
|
||||
"duration": 10.0,
|
||||
"codec": "h264",
|
||||
"width": 1920,
|
||||
"height": 1080,
|
||||
}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_small_file_invalid(self):
|
||||
metadata = {"size_bytes": 100, "duration": 10.0}
|
||||
assert is_valid_media(metadata, "video") is False
|
||||
|
||||
def test_exact_min_size_valid(self):
|
||||
metadata = {"size_bytes": MIN_VIDEO_FILE_SIZE, "duration": 1.0}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_zero_duration_invalid(self):
|
||||
metadata = {"size_bytes": 1024 * 1024, "duration": 0}
|
||||
assert is_valid_media(metadata, "video") is False
|
||||
|
||||
def test_negative_duration_invalid(self):
|
||||
metadata = {"size_bytes": 1024 * 1024, "duration": -1.0}
|
||||
assert is_valid_media(metadata, "video") is False
|
||||
|
||||
def test_unsupported_codec_still_valid(self):
|
||||
# 非白名单编码仍允许通过(不做严格拦截)
|
||||
metadata = {
|
||||
"size_bytes": 1024 * 1024,
|
||||
"duration": 10.0,
|
||||
"codec": "unknown_codec_xyz",
|
||||
}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_empty_codec_valid(self):
|
||||
metadata = {"size_bytes": 1024 * 1024, "duration": 10.0, "codec": ""}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_no_codec_valid(self):
|
||||
metadata = {"size_bytes": 1024 * 1024, "duration": 10.0}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_hevc_codec_valid(self):
|
||||
metadata = {
|
||||
"size_bytes": 1024 * 1024,
|
||||
"duration": 10.0,
|
||||
"codec": "hevc",
|
||||
}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_codec_case_insensitive(self):
|
||||
metadata = {
|
||||
"size_bytes": 1024 * 1024,
|
||||
"duration": 10.0,
|
||||
"codec": "H264",
|
||||
}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_missing_size_invalid(self):
|
||||
metadata = {"duration": 10.0}
|
||||
assert is_valid_media(metadata, "video") is False
|
||||
|
||||
def test_missing_duration_invalid(self):
|
||||
metadata = {"size_bytes": 1024 * 1024}
|
||||
assert is_valid_media(metadata, "video") is False
|
||||
|
||||
def test_empty_metadata_invalid(self):
|
||||
assert is_valid_media({}, "video") is False
|
||||
|
||||
|
||||
# ── is_valid_media - audio ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsValidMediaAudio:
|
||||
"""is_valid_media 音频校验"""
|
||||
|
||||
def test_valid_audio(self):
|
||||
metadata = {"size_bytes": 1024, "duration": 30.0, "codec": "aac"}
|
||||
assert is_valid_media(metadata, "audio") is True
|
||||
|
||||
def test_small_audio_invalid(self):
|
||||
metadata = {"size_bytes": 50, "duration": 30.0}
|
||||
assert is_valid_media(metadata, "audio") is False
|
||||
|
||||
def test_exact_min_size_valid(self):
|
||||
metadata = {"size_bytes": MIN_AUDIO_FILE_SIZE, "duration": 1.0}
|
||||
assert is_valid_media(metadata, "audio") is True
|
||||
|
||||
def test_zero_duration_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "duration": 0}
|
||||
assert is_valid_media(metadata, "audio") is False
|
||||
|
||||
def test_negative_duration_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "duration": -5.0}
|
||||
assert is_valid_media(metadata, "audio") is False
|
||||
|
||||
def test_empty_metadata_invalid(self):
|
||||
assert is_valid_media({}, "audio") is False
|
||||
|
||||
def test_very_short_audio_valid(self):
|
||||
metadata = {"size_bytes": 200, "duration": 0.5}
|
||||
assert is_valid_media(metadata, "audio") is True
|
||||
|
||||
|
||||
# ── is_valid_media - image ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsValidMediaImage:
|
||||
"""is_valid_media 图片校验"""
|
||||
|
||||
def test_valid_image(self):
|
||||
metadata = {"size_bytes": 1024, "width": 1920, "height": 1080}
|
||||
assert is_valid_media(metadata, "image") is True
|
||||
|
||||
def test_small_image_invalid(self):
|
||||
metadata = {"size_bytes": 50, "width": 1920, "height": 1080}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_exact_min_size_valid(self):
|
||||
metadata = {
|
||||
"size_bytes": MIN_IMAGE_FILE_SIZE,
|
||||
"width": 100,
|
||||
"height": 100,
|
||||
}
|
||||
assert is_valid_media(metadata, "image") is True
|
||||
|
||||
def test_zero_width_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "width": 0, "height": 1080}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_zero_height_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "width": 1920, "height": 0}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_negative_width_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "width": -1, "height": 1080}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_small_image_valid(self):
|
||||
metadata = {"size_bytes": 200, "width": 10, "height": 10}
|
||||
assert is_valid_media(metadata, "image") is True
|
||||
|
||||
def test_missing_width_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "height": 1080}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_missing_height_invalid(self):
|
||||
metadata = {"size_bytes": 1024, "width": 1920}
|
||||
assert is_valid_media(metadata, "image") is False
|
||||
|
||||
def test_empty_metadata_invalid(self):
|
||||
assert is_valid_media({}, "image") is False
|
||||
|
||||
|
||||
# ── is_valid_media - edge cases ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsValidMediaEdgeCases:
|
||||
"""is_valid_media 边界情况"""
|
||||
|
||||
def test_invalid_media_type(self):
|
||||
metadata = {"size_bytes": 1024, "duration": 10.0}
|
||||
assert is_valid_media(metadata, "document") is False
|
||||
|
||||
def test_empty_media_type(self):
|
||||
metadata = {"size_bytes": 1024}
|
||||
assert is_valid_media(metadata, "") is False
|
||||
|
||||
def test_string_size_converted(self):
|
||||
metadata = {"size_bytes": "2048", "duration": "5.0"}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_video_size_as_string(self):
|
||||
metadata = {"size_bytes": "1000000", "duration": "30"}
|
||||
assert is_valid_media(metadata, "video") is True
|
||||
|
||||
def test_invalid_size_string_raises(self):
|
||||
# int("abc") 会抛 ValueError
|
||||
metadata = {"size_bytes": "abc", "duration": 10.0}
|
||||
with pytest.raises(ValueError):
|
||||
is_valid_media(metadata, "video")
|
||||
|
||||
def test_none_size_raises(self):
|
||||
# int(None) 会抛 TypeError
|
||||
metadata = {"size_bytes": None, "duration": 10.0}
|
||||
with pytest.raises(TypeError):
|
||||
is_valid_media(metadata, "video")
|
||||
@@ -1,280 +0,0 @@
|
||||
"""VerificationCode 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from domain.verification_code import VerificationCode
|
||||
|
||||
|
||||
class TestVerificationCodeCreate:
|
||||
"""create() 工厂方法测试."""
|
||||
|
||||
def test_create_basic(self):
|
||||
vc = VerificationCode.create("test@example.com", "email_login")
|
||||
assert vc.id is not None
|
||||
assert len(vc.id) == 32
|
||||
assert vc.recipient == "test@example.com"
|
||||
assert vc.code_type == "email_login"
|
||||
assert len(vc.code) == 6
|
||||
assert vc.code.isdigit()
|
||||
assert vc.used_at is None
|
||||
assert vc.attempts == 0
|
||||
assert vc.created_at is not None
|
||||
assert vc.expires_at > vc.created_at
|
||||
|
||||
def test_create_recipient_stripped(self):
|
||||
vc = VerificationCode.create(" test@example.com ", "email_login")
|
||||
assert vc.recipient == "test@example.com"
|
||||
|
||||
def test_create_custom_code(self):
|
||||
vc = VerificationCode.create("test@example.com", "email_login", custom_code="123456")
|
||||
assert vc.code == "123456"
|
||||
|
||||
def test_create_custom_ttl(self):
|
||||
fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
with patch("domain.verification_code.datetime") as mock_dt:
|
||||
mock_dt.now.return_value = fixed_now
|
||||
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
||||
vc = VerificationCode.create("test@example.com", "email_login", ttl_seconds=60)
|
||||
assert vc.expires_at == fixed_now + timedelta(seconds=60)
|
||||
|
||||
def test_create_default_ttl_300(self):
|
||||
fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
with patch("domain.verification_code.datetime") as mock_dt:
|
||||
mock_dt.now.return_value = fixed_now
|
||||
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
||||
vc = VerificationCode.create("test@example.com", "email_login")
|
||||
assert vc.expires_at == fixed_now + timedelta(seconds=300)
|
||||
|
||||
def test_create_unique_ids(self):
|
||||
vc1 = VerificationCode.create("a@b.com", "email_login")
|
||||
vc2 = VerificationCode.create("a@b.com", "email_login")
|
||||
assert vc1.id != vc2.id
|
||||
|
||||
def test_create_unique_codes(self):
|
||||
codes = set()
|
||||
for _ in range(20):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
codes.add(vc.code)
|
||||
# 20个随机6位码几乎肯定不都一样
|
||||
assert len(codes) > 1
|
||||
|
||||
def test_create_phone_recipient(self):
|
||||
vc = VerificationCode.create("13800138000", "phone_login")
|
||||
assert vc.recipient == "13800138000"
|
||||
assert vc.code_type == "phone_login"
|
||||
|
||||
def test_create_all_code_types(self):
|
||||
for ct in ["email_bind", "phone_bind", "email_login", "phone_login", "reset_password"]:
|
||||
vc = VerificationCode.create("test@example.com", ct)
|
||||
assert vc.code_type == ct
|
||||
|
||||
|
||||
class TestVerificationCodeIsExpired:
|
||||
"""is_expired 属性测试."""
|
||||
|
||||
def test_not_expired_future(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=future,
|
||||
)
|
||||
assert vc.is_expired is False
|
||||
|
||||
def test_expired_past(self):
|
||||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=past,
|
||||
)
|
||||
assert vc.is_expired is True
|
||||
|
||||
def test_expired_boundary_exact(self):
|
||||
# 用mock固定时间,expires_at等于当前时间不算过期
|
||||
fixed_now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
with patch("domain.verification_code.datetime") as mock_dt:
|
||||
mock_dt.now.return_value = fixed_now
|
||||
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=fixed_now,
|
||||
)
|
||||
assert vc.is_expired is False
|
||||
|
||||
|
||||
class TestVerificationCodeIsUsed:
|
||||
"""is_used 属性测试."""
|
||||
|
||||
def test_not_used_default(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
assert vc.is_used is False
|
||||
|
||||
def test_is_used_after_mark(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
vc.mark_used()
|
||||
assert vc.is_used is True
|
||||
|
||||
|
||||
class TestVerificationCodeIsValid:
|
||||
"""is_valid 属性测试."""
|
||||
|
||||
def test_valid_fresh(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=future,
|
||||
)
|
||||
assert vc.is_valid is True
|
||||
|
||||
def test_invalid_expired(self):
|
||||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=past,
|
||||
)
|
||||
assert vc.is_valid is False
|
||||
|
||||
def test_invalid_used(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=future,
|
||||
)
|
||||
vc.mark_used()
|
||||
assert vc.is_valid is False
|
||||
|
||||
def test_invalid_expired_and_used(self):
|
||||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
vc = VerificationCode(
|
||||
id="1",
|
||||
recipient="a@b.com",
|
||||
code="123456",
|
||||
code_type="email_login",
|
||||
expires_at=past,
|
||||
)
|
||||
vc.mark_used()
|
||||
assert vc.is_valid is False
|
||||
|
||||
|
||||
class TestVerificationCodeMarkUsed:
|
||||
"""mark_used 方法测试."""
|
||||
|
||||
def test_mark_used_sets_timestamp(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
assert vc.used_at is None
|
||||
before = datetime.now(timezone.utc)
|
||||
vc.mark_used()
|
||||
after = datetime.now(timezone.utc)
|
||||
assert vc.used_at is not None
|
||||
assert before <= vc.used_at <= after
|
||||
|
||||
def test_mark_used_twice_overwrites(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
vc.mark_used()
|
||||
first = vc.used_at
|
||||
# 时间足够短,一般不会不同,但确保可以重复调用
|
||||
vc.mark_used()
|
||||
assert vc.used_at is not None
|
||||
|
||||
|
||||
class TestVerificationCodeIncrementAttempts:
|
||||
"""increment_attempts 方法测试."""
|
||||
|
||||
def test_default_zero(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
assert vc.attempts == 0
|
||||
|
||||
def test_increment_once(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
vc.increment_attempts()
|
||||
assert vc.attempts == 1
|
||||
|
||||
def test_increment_multiple(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
for _i in range(5):
|
||||
vc.increment_attempts()
|
||||
assert vc.attempts == 5
|
||||
|
||||
|
||||
class TestVerificationCodeBasics:
|
||||
"""基础构造和 slots 测试."""
|
||||
|
||||
def test_direct_construction(self):
|
||||
now = datetime.now(timezone.utc)
|
||||
vc = VerificationCode(
|
||||
id="abc123",
|
||||
recipient="test@test.com",
|
||||
code="000000",
|
||||
code_type="email_bind",
|
||||
expires_at=now + timedelta(minutes=5),
|
||||
used_at=None,
|
||||
attempts=0,
|
||||
created_at=now,
|
||||
)
|
||||
assert vc.id == "abc123"
|
||||
assert vc.recipient == "test@test.com"
|
||||
assert vc.code == "000000"
|
||||
|
||||
def test_slots_no_extra_attrs(self):
|
||||
vc = VerificationCode.create("a@b.com", "email_login")
|
||||
with pytest.raises((AttributeError, TypeError)):
|
||||
vc.new_field = "value" # type: ignore[attr-defined]
|
||||
|
||||
def test_equality_same_id(self):
|
||||
now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
vc1 = VerificationCode(
|
||||
id="same",
|
||||
recipient="a@b.com",
|
||||
code="111",
|
||||
code_type="email_login",
|
||||
expires_at=now,
|
||||
created_at=now,
|
||||
)
|
||||
vc2 = VerificationCode(
|
||||
id="same",
|
||||
recipient="a@b.com",
|
||||
code="111",
|
||||
code_type="email_login",
|
||||
expires_at=now,
|
||||
created_at=now,
|
||||
)
|
||||
assert vc1 == vc2
|
||||
|
||||
def test_equality_different_id(self):
|
||||
now = datetime.now(timezone.utc)
|
||||
vc1 = VerificationCode(
|
||||
id="id1",
|
||||
recipient="a@b.com",
|
||||
code="111",
|
||||
code_type="email_login",
|
||||
expires_at=now,
|
||||
)
|
||||
vc2 = VerificationCode(
|
||||
id="id2",
|
||||
recipient="a@b.com",
|
||||
code="111",
|
||||
code_type="email_login",
|
||||
expires_at=now,
|
||||
)
|
||||
assert vc1 != vc2
|
||||
Reference in New Issue
Block a user