Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 13a1070fbc | |||
| 7f7a66899b |
Executable
+159
@@ -0,0 +1,159 @@
|
||||
"""Auth ports (ABC接口) 单元测试.
|
||||
|
||||
验证抽象接口定义正确:不能直接实例化,子类必须实现所有抽象方法。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC
|
||||
|
||||
import pytest
|
||||
from domain.auth.email_service import EmailServicePort
|
||||
from domain.auth.jwt_service import JWTServicePort
|
||||
from domain.auth.password_hasher import PasswordHasherPort, PasswordValidatorPort
|
||||
from domain.auth.session_store import SessionStorePort
|
||||
from domain.auth.sms_service import SmsService
|
||||
|
||||
|
||||
class TestSessionStorePort:
|
||||
"""SessionStorePort 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(SessionStorePort, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
SessionStorePort() # type: ignore[misc]
|
||||
|
||||
def test_has_abstract_methods(self):
|
||||
abstract_methods = SessionStorePort.__abstractmethods__
|
||||
expected = {
|
||||
"save_session",
|
||||
"get_session",
|
||||
"get_session_by_refresh_token",
|
||||
"get_refresh_token",
|
||||
"update_last_active",
|
||||
"delete_session",
|
||||
"get_user_sessions",
|
||||
"delete_all_user_sessions",
|
||||
"session_exists",
|
||||
}
|
||||
assert expected.issubset(abstract_methods)
|
||||
|
||||
def test_concrete_subclass_works(self):
|
||||
class ConcreteStore(SessionStorePort):
|
||||
def save_session(self, **kwargs): # type: ignore[override]
|
||||
return True
|
||||
|
||||
def get_session(self, session_id): # type: ignore[override]
|
||||
return None
|
||||
|
||||
def get_session_by_refresh_token(self, token): # type: ignore[override]
|
||||
return None
|
||||
|
||||
def get_refresh_token(self, session_id): # type: ignore[override]
|
||||
return None
|
||||
|
||||
def update_last_active(self, session_id): # type: ignore[override]
|
||||
return True
|
||||
|
||||
def delete_session(self, session_id): # type: ignore[override]
|
||||
return True
|
||||
|
||||
def get_user_sessions(self, user_id): # type: ignore[override]
|
||||
return []
|
||||
|
||||
def delete_all_user_sessions(self, user_id): # type: ignore[override]
|
||||
return 0
|
||||
|
||||
def session_exists(self, session_id): # type: ignore[override]
|
||||
return False
|
||||
|
||||
store = ConcreteStore()
|
||||
assert isinstance(store, SessionStorePort)
|
||||
assert store.session_exists("s1") is False
|
||||
assert store.delete_session("s1") is True
|
||||
|
||||
|
||||
class TestEmailServicePort:
|
||||
"""EmailServicePort 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(EmailServicePort, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
EmailServicePort() # type: ignore[misc]
|
||||
|
||||
def test_has_abstract_methods(self):
|
||||
abstract_methods = EmailServicePort.__abstractmethods__
|
||||
expected = {"send_email", "send_verification_email", "send_password_reset_email"}
|
||||
assert expected.issubset(abstract_methods)
|
||||
|
||||
|
||||
class TestJWTServicePort:
|
||||
"""JWTServicePort 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(JWTServicePort, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
JWTServicePort() # type: ignore[misc]
|
||||
|
||||
def test_has_abstract_methods(self):
|
||||
abstract_methods = JWTServicePort.__abstractmethods__
|
||||
expected = {
|
||||
"create_access_token",
|
||||
"create_refresh_token",
|
||||
"verify_token",
|
||||
"verify_access_token",
|
||||
"verify_refresh_token",
|
||||
}
|
||||
assert expected.issubset(abstract_methods)
|
||||
|
||||
|
||||
class TestPasswordHasherPort:
|
||||
"""PasswordHasherPort 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(PasswordHasherPort, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
PasswordHasherPort() # type: ignore[misc]
|
||||
|
||||
def test_has_abstract_methods(self):
|
||||
abstract_methods = PasswordHasherPort.__abstractmethods__
|
||||
expected = {"hash_password", "verify_password", "needs_rehash"}
|
||||
assert expected.issubset(abstract_methods)
|
||||
|
||||
|
||||
class TestPasswordValidatorPort:
|
||||
"""PasswordValidatorPort 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(PasswordValidatorPort, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
PasswordValidatorPort() # type: ignore[misc]
|
||||
|
||||
def test_has_validate_method(self):
|
||||
assert "validate" in PasswordValidatorPort.__abstractmethods__
|
||||
|
||||
|
||||
class TestSmsService:
|
||||
"""SmsService 接口测试."""
|
||||
|
||||
def test_is_abstract(self):
|
||||
assert issubclass(SmsService, ABC)
|
||||
|
||||
def test_cannot_instantiate(self):
|
||||
with pytest.raises(TypeError):
|
||||
SmsService() # type: ignore[misc]
|
||||
|
||||
def test_has_abstract_methods(self):
|
||||
abstract_methods = SmsService.__abstractmethods__
|
||||
expected = {"send_verification_code", "send_template_sms"}
|
||||
assert expected.issubset(abstract_methods)
|
||||
Executable
+139
@@ -0,0 +1,139 @@
|
||||
"""bgm_utils 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from domain.bgm_utils import merge_bgm_config
|
||||
|
||||
|
||||
class TestMergeBgmConfigEmptyInputs:
|
||||
"""空输入测试."""
|
||||
|
||||
def test_both_empty(self):
|
||||
result = merge_bgm_config({}, {})
|
||||
assert result == {}
|
||||
|
||||
def test_user_empty_returns_template_copy(self):
|
||||
template = {"enabled": True, "volume": 0.5}
|
||||
result = merge_bgm_config(template, {})
|
||||
assert result == {"enabled": True, "volume": 0.5}
|
||||
# 返回的是副本不是同一个对象
|
||||
assert result is not template
|
||||
|
||||
def test_template_empty_returns_user_copy(self):
|
||||
user = {"enabled": False, "volume": 0.8}
|
||||
result = merge_bgm_config({}, user)
|
||||
assert result == {"enabled": False, "volume": 0.8}
|
||||
assert result is not user
|
||||
|
||||
def test_user_none_returns_template(self):
|
||||
template = {"enabled": True}
|
||||
result = merge_bgm_config(template, None) # type: ignore[arg-type]
|
||||
assert result == template
|
||||
|
||||
def test_template_none_returns_user(self):
|
||||
user = {"enabled": True}
|
||||
result = merge_bgm_config(None, user) # type: ignore[arg-type]
|
||||
assert result == user
|
||||
|
||||
|
||||
class TestMergeBgmConfigBasicMerge:
|
||||
"""基础合并测试."""
|
||||
|
||||
def test_user_overrides_template_field(self):
|
||||
template = {"volume": 0.5, "fade_in": 1.0}
|
||||
user = {"volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["volume"] == 0.8
|
||||
assert result["fade_in"] == 1.0
|
||||
|
||||
def test_user_adds_new_field(self):
|
||||
template = {"volume": 0.5}
|
||||
user = {"fade_out": 2.0}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["volume"] == 0.5
|
||||
assert result["fade_out"] == 2.0
|
||||
|
||||
def test_all_fields_overridden(self):
|
||||
template = {"enabled": True, "volume": 0.5, "track_id": "t1"}
|
||||
user = {"enabled": False, "volume": 1.0, "track_id": "t2"}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result == {"enabled": False, "volume": 1.0, "track_id": "t2"}
|
||||
|
||||
|
||||
class TestMergeBgmConfigEnabledSpecial:
|
||||
"""enabled 特殊处理测试."""
|
||||
|
||||
def test_user_no_enabled_keeps_template_enabled_true(self):
|
||||
template = {"enabled": True, "volume": 0.5}
|
||||
user = {"volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["enabled"] is True
|
||||
assert result["volume"] == 0.8
|
||||
|
||||
def test_user_no_enabled_keeps_template_enabled_false(self):
|
||||
template = {"enabled": False, "volume": 0.5}
|
||||
user = {"volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["enabled"] is False
|
||||
assert result["volume"] == 0.8
|
||||
|
||||
def test_user_explicit_enabled_true_overrides_template_false(self):
|
||||
template = {"enabled": False, "volume": 0.5}
|
||||
user = {"enabled": True, "volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["enabled"] is True
|
||||
assert result["volume"] == 0.8
|
||||
|
||||
def test_user_explicit_enabled_false_overrides_template_true(self):
|
||||
template = {"enabled": True, "volume": 0.5}
|
||||
user = {"enabled": False}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["enabled"] is False
|
||||
|
||||
def test_template_no_enabled_user_no_enabled(self):
|
||||
template = {"volume": 0.5}
|
||||
user = {"volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert "enabled" not in result
|
||||
assert result["volume"] == 0.8
|
||||
|
||||
def test_template_no_enabled_user_has_enabled(self):
|
||||
template = {"volume": 0.5}
|
||||
user = {"enabled": True, "volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["enabled"] is True
|
||||
|
||||
|
||||
class TestMergeBgmConfigDoesNotMutate:
|
||||
"""不修改原字典测试."""
|
||||
|
||||
def test_template_not_mutated(self):
|
||||
template = {"enabled": True, "volume": 0.5}
|
||||
original = dict(template)
|
||||
user = {"volume": 0.8, "fade": 1.0}
|
||||
merge_bgm_config(template, user)
|
||||
assert template == original
|
||||
|
||||
def test_user_not_mutated(self):
|
||||
template = {"enabled": True, "volume": 0.5}
|
||||
user = {"volume": 0.8}
|
||||
original = dict(user)
|
||||
merge_bgm_config(template, user)
|
||||
assert user == original
|
||||
|
||||
|
||||
class TestMergeBgmConfigNestedDict:
|
||||
"""嵌套字典合并测试(简单合并,非深合并)."""
|
||||
|
||||
def test_nested_dict_user_overrides(self):
|
||||
template = {"effects": {"fade_in": 1.0, "fade_out": 1.0}}
|
||||
user = {"effects": {"fade_in": 2.0}}
|
||||
result = merge_bgm_config(template, user)
|
||||
# 简单合并,用户effects整个覆盖模板的
|
||||
assert result["effects"] == {"fade_in": 2.0}
|
||||
|
||||
def test_nested_dict_preserved_when_no_user_override(self):
|
||||
template = {"effects": {"fade_in": 1.0}}
|
||||
user = {"volume": 0.8}
|
||||
result = merge_bgm_config(template, user)
|
||||
assert result["effects"] == {"fade_in": 1.0}
|
||||
Executable
+63
@@ -0,0 +1,63 @@
|
||||
"""EmailConfig 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from domain.auth.email_service import EmailConfig
|
||||
|
||||
|
||||
class TestEmailConfigDefaults:
|
||||
"""默认值测试."""
|
||||
|
||||
def test_default_values(self):
|
||||
config = EmailConfig()
|
||||
assert config.smtp_host == "smtp.gmail.com"
|
||||
assert config.smtp_port == 587
|
||||
assert config.smtp_user == ""
|
||||
assert config.smtp_password == ""
|
||||
assert config.from_email == ""
|
||||
assert config.from_name == "小虾 SaaS"
|
||||
assert config.use_tls is True
|
||||
|
||||
def test_custom_construction(self):
|
||||
config = EmailConfig(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=465,
|
||||
smtp_user="user@example.com",
|
||||
smtp_password="secret",
|
||||
from_email="no-reply@example.com",
|
||||
from_name="Example App",
|
||||
use_tls=False,
|
||||
)
|
||||
assert config.smtp_host == "smtp.example.com"
|
||||
assert config.smtp_port == 465
|
||||
assert config.smtp_user == "user@example.com"
|
||||
assert config.smtp_password == "secret"
|
||||
assert config.from_email == "no-reply@example.com"
|
||||
assert config.from_name == "Example App"
|
||||
assert config.use_tls is False
|
||||
|
||||
def test_is_dataclass(self):
|
||||
# 可重复创建相同配置
|
||||
c1 = EmailConfig(smtp_host="h.com", smtp_port=25)
|
||||
c2 = EmailConfig(smtp_host="h.com", smtp_port=25)
|
||||
assert c1 == c2
|
||||
|
||||
|
||||
class TestEmailConfigEquality:
|
||||
"""相等性测试."""
|
||||
|
||||
def test_equal_same_values(self):
|
||||
c1 = EmailConfig()
|
||||
c2 = EmailConfig()
|
||||
assert c1 == c2
|
||||
|
||||
def test_not_equal_different_host(self):
|
||||
c1 = EmailConfig(smtp_host="a.com")
|
||||
c2 = EmailConfig(smtp_host="b.com")
|
||||
assert c1 != c2
|
||||
|
||||
def test_not_equal_different_port(self):
|
||||
c1 = EmailConfig(smtp_port=587)
|
||||
c2 = EmailConfig(smtp_port=465)
|
||||
assert c1 != c2
|
||||
Executable
+107
@@ -0,0 +1,107 @@
|
||||
"""domain exceptions 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from domain.exceptions import (
|
||||
DomainError,
|
||||
NotFoundError,
|
||||
QuotaExceededError,
|
||||
ValidationError,
|
||||
)
|
||||
|
||||
|
||||
class TestDomainError:
|
||||
"""DomainError 基类测试."""
|
||||
|
||||
def test_is_exception(self):
|
||||
assert issubclass(DomainError, Exception)
|
||||
|
||||
def test_raise_and_catch(self):
|
||||
with pytest.raises(DomainError):
|
||||
raise DomainError("something went wrong")
|
||||
|
||||
def test_message(self):
|
||||
err = DomainError("test message")
|
||||
assert str(err) == "test message"
|
||||
|
||||
def test_empty_message(self):
|
||||
err = DomainError()
|
||||
assert str(err) == ""
|
||||
|
||||
|
||||
class TestNotFoundError:
|
||||
"""NotFoundError 测试."""
|
||||
|
||||
def test_is_domain_error(self):
|
||||
assert issubclass(NotFoundError, DomainError)
|
||||
|
||||
def test_raise_and_catch_as_domain(self):
|
||||
with pytest.raises(DomainError):
|
||||
raise NotFoundError("user not found")
|
||||
|
||||
def test_raise_and_catch_specific(self):
|
||||
with pytest.raises(NotFoundError):
|
||||
raise NotFoundError("user not found")
|
||||
|
||||
def test_message(self):
|
||||
err = NotFoundError("resource not found")
|
||||
assert str(err) == "resource not found"
|
||||
|
||||
|
||||
class TestValidationError:
|
||||
"""ValidationError 测试."""
|
||||
|
||||
def test_is_domain_error(self):
|
||||
assert issubclass(ValidationError, DomainError)
|
||||
|
||||
def test_raise_and_catch_as_domain(self):
|
||||
with pytest.raises(DomainError):
|
||||
raise ValidationError("invalid input")
|
||||
|
||||
def test_raise_and_catch_specific(self):
|
||||
with pytest.raises(ValidationError):
|
||||
raise ValidationError("invalid input")
|
||||
|
||||
def test_message(self):
|
||||
err = ValidationError("bad data")
|
||||
assert str(err) == "bad data"
|
||||
|
||||
|
||||
class TestQuotaExceededError:
|
||||
"""QuotaExceededError 测试."""
|
||||
|
||||
def test_is_domain_error(self):
|
||||
assert issubclass(QuotaExceededError, DomainError)
|
||||
|
||||
def test_constructor_stores_fields(self):
|
||||
err = QuotaExceededError("storage", limit=100.0, used=150.0)
|
||||
assert err.dimension == "storage"
|
||||
assert err.limit == 100.0
|
||||
assert err.used == 150.0
|
||||
|
||||
def test_message_format(self):
|
||||
err = QuotaExceededError("storage", limit=100.0, used=150.0)
|
||||
assert "storage" in str(err)
|
||||
assert "150.0" in str(err)
|
||||
assert "100.0" in str(err)
|
||||
assert "Quota exceeded" in str(err)
|
||||
|
||||
def test_raise_and_catch_as_domain(self):
|
||||
with pytest.raises(DomainError):
|
||||
raise QuotaExceededError("api_calls", limit=1000, used=2000)
|
||||
|
||||
def test_raise_and_catch_specific(self):
|
||||
with pytest.raises(QuotaExceededError):
|
||||
raise QuotaExceededError("api_calls", limit=1000, used=2000)
|
||||
|
||||
def test_int_values(self):
|
||||
err = QuotaExceededError("count", limit=100, used=150)
|
||||
assert err.limit == 100
|
||||
assert err.used == 150
|
||||
assert "150/100" in str(err)
|
||||
|
||||
def test_float_values(self):
|
||||
err = QuotaExceededError("size", limit=10.5, used=20.3)
|
||||
assert err.limit == 10.5
|
||||
assert err.used == 20.3
|
||||
Executable
+134
@@ -0,0 +1,134 @@
|
||||
"""EditTemplateVersion 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from domain.template_version import EditTemplateVersion
|
||||
|
||||
|
||||
class TestEditTemplateVersionCreate:
|
||||
"""create() 工厂方法测试."""
|
||||
|
||||
def test_create_basic(self):
|
||||
v = EditTemplateVersion.create("tmpl_001", 1)
|
||||
assert v.id is not None
|
||||
assert len(v.id) == 32
|
||||
assert v.template_id == "tmpl_001"
|
||||
assert v.version == 1
|
||||
assert v.name == ""
|
||||
assert v.editing_mode == "one_take"
|
||||
assert v.config == {}
|
||||
assert v.clip_configs == []
|
||||
assert v.change_note == ""
|
||||
assert v.published_by == ""
|
||||
assert v.created_at is not None
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
v = EditTemplateVersion.create(
|
||||
"tmpl_001",
|
||||
3,
|
||||
name="第三版",
|
||||
editing_mode="pip",
|
||||
config={"bgm": True},
|
||||
clip_configs=[{"id": "c1", "type": "video"}],
|
||||
change_note="优化剪辑逻辑",
|
||||
published_by="user_123",
|
||||
)
|
||||
assert v.template_id == "tmpl_001"
|
||||
assert v.version == 3
|
||||
assert v.name == "第三版"
|
||||
assert v.editing_mode == "pip"
|
||||
assert v.config == {"bgm": True}
|
||||
assert v.clip_configs == [{"id": "c1", "type": "video"}]
|
||||
assert v.change_note == "优化剪辑逻辑"
|
||||
assert v.published_by == "user_123"
|
||||
|
||||
def test_create_config_none_defaults_to_empty(self):
|
||||
v = EditTemplateVersion.create("t1", 1, config=None)
|
||||
assert v.config == {}
|
||||
|
||||
def test_create_clip_configs_none_defaults_to_empty(self):
|
||||
v = EditTemplateVersion.create("t1", 1, clip_configs=None)
|
||||
assert v.clip_configs == []
|
||||
|
||||
def test_create_unique_ids(self):
|
||||
v1 = EditTemplateVersion.create("t1", 1)
|
||||
v2 = EditTemplateVersion.create("t1", 1)
|
||||
assert v1.id != v2.id
|
||||
|
||||
def test_create_sets_created_at(self):
|
||||
before = datetime.now(timezone.utc)
|
||||
v = EditTemplateVersion.create("t1", 1)
|
||||
after = datetime.now(timezone.utc)
|
||||
assert before <= v.created_at <= after
|
||||
|
||||
|
||||
class TestEditTemplateVersionConstruction:
|
||||
"""直接构造测试."""
|
||||
|
||||
def test_direct_construction(self):
|
||||
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
v = EditTemplateVersion(
|
||||
id="v1",
|
||||
template_id="t1",
|
||||
version=5,
|
||||
name="v5",
|
||||
editing_mode="voice_over",
|
||||
config={"key": "value"},
|
||||
clip_configs=[{"a": 1}, {"b": 2}],
|
||||
change_note="test",
|
||||
published_by="admin",
|
||||
created_at=now,
|
||||
)
|
||||
assert v.id == "v1"
|
||||
assert v.template_id == "t1"
|
||||
assert v.version == 5
|
||||
assert v.name == "v5"
|
||||
assert v.editing_mode == "voice_over"
|
||||
assert v.config == {"key": "value"}
|
||||
assert v.clip_configs == [{"a": 1}, {"b": 2}]
|
||||
assert v.change_note == "test"
|
||||
assert v.published_by == "admin"
|
||||
assert v.created_at == now
|
||||
|
||||
def test_default_values(self):
|
||||
v = EditTemplateVersion(id="v1", template_id="t1", version=1)
|
||||
assert v.name == ""
|
||||
assert v.editing_mode == "one_take"
|
||||
assert v.config == {}
|
||||
assert v.clip_configs == []
|
||||
assert v.change_note == ""
|
||||
assert v.published_by == ""
|
||||
|
||||
|
||||
class TestEditTemplateVersionSlots:
|
||||
"""slots 测试."""
|
||||
|
||||
def test_slots_no_extra_attrs(self):
|
||||
v = EditTemplateVersion.create("t1", 1)
|
||||
with pytest.raises((AttributeError, TypeError)):
|
||||
v.new_field = "value" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class TestEditTemplateVersionEquality:
|
||||
"""相等性测试."""
|
||||
|
||||
def test_equal_same_id_and_version(self):
|
||||
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
v1 = EditTemplateVersion(id="same", template_id="t1", version=1, created_at=now)
|
||||
v2 = EditTemplateVersion(id="same", template_id="t1", version=1, created_at=now)
|
||||
assert v1 == v2
|
||||
|
||||
def test_not_equal_different_id(self):
|
||||
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
v1 = EditTemplateVersion(id="v1", template_id="t1", version=1, created_at=now)
|
||||
v2 = EditTemplateVersion(id="v2", template_id="t1", version=1, created_at=now)
|
||||
assert v1 != v2
|
||||
|
||||
def test_not_equal_different_version(self):
|
||||
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
v1 = EditTemplateVersion(id="same", template_id="t1", version=1, created_at=now)
|
||||
v2 = EditTemplateVersion(id="same", template_id="t1", version=2, created_at=now)
|
||||
assert v1 != v2
|
||||
@@ -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