"""绑定联系方式 Use Case 单元测试.""" from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock import pytest from packages.application.auth.bind_contact_use_case import ( BindContactRequest, BindContactResponse, BindContactUseCase, SendVerificationCodeRequest, SendVerificationCodeResponse, SendVerificationCodeUseCase, ) from packages.domain.verification_code import VerificationCode # ── Test Fixtures ──────────────────────────────────────────────────────────── def _make_user( user_id="user-1", email="wechat_user@wechat.local", username="", display_name="微信用户", phone=None, phone_verified=False, email_verified=False, binding_completed_at=None, ): """创建测试用户.""" from packages.domain.entities import User user = User( id=user_id, email=email, display_name=display_name, username=username, ) user.phone = phone user.phone_verified = phone_verified user.email_verified = email_verified user.binding_completed_at = binding_completed_at return user @pytest.fixture def mock_user_repo(): repo = MagicMock() repo.find_by_id.return_value = None repo.find_by_phone.return_value = None repo.find_by_email.return_value = None repo.save.return_value = None return repo @pytest.fixture def mock_verification_service(): svc = MagicMock() svc.verify.return_value = (True, None) svc.generate.return_value = (None, None) return svc @pytest.fixture def mock_email_service(): return MagicMock() @pytest.fixture def mock_sms_service(): return MagicMock() # ── BindContactRequest 测试 ───────────────────────────────────────────────── class TestBindContactRequest: """绑定请求数据类测试""" def test_phone_normalized(self): """手机号被标准化.""" req = BindContactRequest(user_id="u1", phone="+8613800138000", phone_code="123456") assert req.phone == "13800138000" def test_phone_code_stripped(self): """手机验证码去除前后空格.""" req = BindContactRequest(user_id="u1", phone="13800138000", phone_code=" 123 ") assert req.phone_code == "123" def test_email_normalized(self): """邮箱被小写化+去空格.""" req = BindContactRequest(user_id="u1", email=" User@Example.COM ", email_code="abcdef") assert req.email == "user@example.com" def test_email_code_stripped(self): """邮箱验证码去除前后空格.""" req = BindContactRequest(user_id="u1", email="u@e.com", email_code=" abc ") assert req.email_code == "abc" def test_empty_phone_stays_empty(self): """空手机号保持空.""" req = BindContactRequest(user_id="u1", phone="", email="u@e.com") assert req.phone == "" def test_empty_email_stays_empty(self): """空邮箱保持空.""" req = BindContactRequest(user_id="u1", phone="13800138000", email="") assert req.email == "" def test_user_id_preserved(self): """user_id保持不变.""" req = BindContactRequest(user_id="user-abc-123", phone="13800138000") assert req.user_id == "user-abc-123" # ── BindContactResponse 测试 ──────────────────────────────────────────────── class TestBindContactResponse: """绑定响应数据类测试""" def test_to_dict_structure(self): """to_dict 返回正确结构.""" user = _make_user(email="user@e.com", phone="13800138000", phone_verified=True) resp = BindContactResponse(user=user) d = resp.to_dict() assert "user" in d u = d["user"] assert u["id"] == "user-1" assert u["email"] == "user@e.com" assert u["phone"] == "13800138000" assert u["phone_verified"] is True assert u["display_name"] == "微信用户" assert u["binding_complete"] is False def test_to_dict_binding_complete(self): """绑定完成时 binding_complete 为 True.""" user = _make_user( email="user@e.com", phone="13800138000", phone_verified=True, email_verified=True, binding_completed_at=datetime.now(timezone.utc), ) resp = BindContactResponse(user=user) assert resp.to_dict()["user"]["binding_complete"] is True # ── BindContactUseCase 手机绑定测试 ──────────────────────────────────────── class TestBindContactPhone: """手机绑定测试""" def test_bind_phone_success(self, mock_user_repo, mock_verification_service): """成功绑定手机号.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456") resp, error = uc.execute(req) assert error is None assert resp is not None assert resp.user.phone == "13800138000" assert resp.user.phone_verified is True mock_user_repo.save.assert_called_once() mock_verification_service.verify.assert_called_once() def test_bind_phone_invalid_format(self, mock_user_repo, mock_verification_service): """手机号格式错误.""" user = _make_user() mock_user_repo.find_by_id.return_value = user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="123", phone_code="123456") resp, error = uc.execute(req) assert resp is None assert "格式" in error or "不正确" in error def test_bind_phone_missing_code(self, mock_user_repo, mock_verification_service): """缺少手机验证码.""" user = _make_user() mock_user_repo.find_by_id.return_value = user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="") resp, error = uc.execute(req) assert resp is None assert "请输入手机验证码" in error def test_bind_phone_already_used(self, mock_user_repo, mock_verification_service): """手机号已被其他账号绑定.""" from packages.domain.entities import User user = _make_user() other_user = User(id="user-2", email="o@e.com", display_name="Other") other_user.phone = "13800138000" mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = other_user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456") resp, error = uc.execute(req) assert resp is None assert "已被其他账号绑定" in error def test_bind_phone_same_user_allowed(self, mock_user_repo, mock_verification_service): """同一用户绑定自己已有的手机号（重新验证）.""" user = _make_user(phone="13800138000", phone_verified=True) mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = user # 同一个用户 mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456") resp, error = uc.execute(req) assert error is None assert resp is not None assert resp.user.phone == "13800138000" def test_bind_phone_wrong_code(self, mock_user_repo, mock_verification_service): """手机验证码错误.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_verification_service.verify.return_value = (False, "验证码错误") uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="000000") resp, error = uc.execute(req) assert resp is None assert "手机验证码错误" in error def test_bind_phone_normalized(self, mock_user_repo, mock_verification_service): """+86前缀的手机号被标准化后再验证.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) # BindContactRequest 会自动标准化 +86 req = BindContactRequest(user_id="user-1", phone="+8613800138000", phone_code="123456") resp, error = uc.execute(req) assert error is None assert resp.user.phone == "13800138000" # 标准化后的号码 # ── BindContactUseCase 邮箱绑定测试 ──────────────────────────────────────── class TestBindContactEmail: """邮箱绑定测试""" def test_bind_email_success(self, mock_user_repo, mock_verification_service): """成功绑定邮箱.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_email.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", email="new@example.com", email_code="abcdef") resp, error = uc.execute(req) assert error is None assert resp is not None assert resp.user.email == "new@example.com" assert resp.user.email_verified is True mock_user_repo.save.assert_called_once() def test_bind_email_invalid_format(self, mock_user_repo, mock_verification_service): """邮箱格式错误.""" user = _make_user() mock_user_repo.find_by_id.return_value = user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", email="invalid", email_code="abc") resp, error = uc.execute(req) assert resp is None assert "格式" in error or "不正确" in error def test_bind_email_missing_code(self, mock_user_repo, mock_verification_service): """缺少邮箱验证码.""" user = _make_user() mock_user_repo.find_by_id.return_value = user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", email="u@e.com", email_code="") resp, error = uc.execute(req) assert resp is None assert "请输入邮箱验证码" in error def test_bind_email_already_used(self, mock_user_repo, mock_verification_service): """邮箱已被其他账号绑定.""" from packages.domain.entities import User user = _make_user() other_user = User(id="user-2", email="taken@e.com", display_name="Other") mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_email.return_value = other_user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", email="taken@e.com", email_code="abcdef") resp, error = uc.execute(req) assert resp is None assert "已被其他账号绑定" in error def test_bind_email_wrong_code(self, mock_user_repo, mock_verification_service): """邮箱验证码错误.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_email.return_value = None mock_verification_service.verify.return_value = (False, "验证码错误") uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", email="u@e.com", email_code="wrongcode") resp, error = uc.execute(req) assert resp is None assert "邮箱验证码错误" in error # ── BindContactUseCase 组合绑定测试 ──────────────────────────────────────── class TestBindContactCombined: """手机+邮箱组合绑定测试""" def test_bind_both_phone_and_email(self, mock_user_repo, mock_verification_service): """同时绑定手机和邮箱.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_user_repo.find_by_email.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest( user_id="user-1", phone="13800138000", phone_code="123456", email="user@e.com", email_code="abcdef", ) resp, error = uc.execute(req) assert error is None assert resp is not None assert resp.user.phone == "13800138000" assert resp.user.email == "user@e.com" assert resp.user.phone_verified is True assert resp.user.email_verified is True # 两者都验证通过且邮箱不是wechat.local，应该标记绑定完成 assert resp.user.binding_completed_at is not None def test_bind_both_phone_fails_email_not_attempted(self, mock_user_repo, mock_verification_service): """手机验证失败时不继续验证邮箱.""" user = _make_user() mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_verification_service.verify.return_value = (False, "验证码错误") uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest( user_id="user-1", phone="13800138000", phone_code="wrong", email="u@e.com", email_code="abcdef", ) resp, error = uc.execute(req) assert resp is None # verify 应该只被调用了一次（手机验证失败就返回了） assert mock_verification_service.verify.call_count == 1 def test_binding_complete_only_with_real_email(self, mock_user_repo, mock_verification_service): """只有邮箱不是wechat.local时才算绑定完成.""" # 用户已有 @wechat.local 邮箱（微信登录的默认邮箱），只绑定手机 user = _make_user( email="wx_123@wechat.local", phone="13800138000", phone_verified=True, ) mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456") resp, _ = uc.execute(req) # 只有手机验证通过，但邮箱还是wechat.local，不算绑定完成 assert resp.user.binding_completed_at is None def test_binding_complete_with_both_verified(self, mock_user_repo, mock_verification_service): """手机+真实邮箱都验证通过后，标记绑定完成.""" user = _make_user( email="wx_123@wechat.local", phone="13800138000", phone_verified=True, ) mock_user_repo.find_by_id.return_value = user mock_user_repo.find_by_phone.return_value = None mock_user_repo.find_by_email.return_value = None mock_verification_service.verify.return_value = (True, None) uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest( user_id="user-1", phone="13800138000", phone_code="123456", email="real@e.com", email_code="abcdef", ) resp, _ = uc.execute(req) assert resp.user.binding_completed_at is not None assert resp.user.email == "real@e.com" # ── BindContactUseCase 通用/边界测试 ────────────────────────────────────── class TestBindContactGeneral: """绑定通用/边界测试""" def test_no_phone_no_email_error(self, mock_user_repo, mock_verification_service): """手机和邮箱都没填.""" user = _make_user() mock_user_repo.find_by_id.return_value = user uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="user-1") resp, error = uc.execute(req) assert resp is None assert "至少填写" in error def test_user_not_found(self, mock_user_repo, mock_verification_service): """用户不存在.""" mock_user_repo.find_by_id.return_value = None uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="nonexistent", phone="13800138000") resp, error = uc.execute(req) assert resp is None assert "用户不存在" in error def test_exception_returns_error(self, mock_user_repo, mock_verification_service): """异常情况返回错误.""" mock_user_repo.find_by_id.side_effect = Exception("DB error") uc = BindContactUseCase( user_repository=mock_user_repo, verification_code_service=mock_verification_service, ) req = BindContactRequest(user_id="u1", phone="13800138000") resp, error = uc.execute(req) assert resp is None assert "失败" in error # ── SendVerificationCodeRequest 测试 ─────────────────────────────────────── class TestSendVerificationCodeRequest: """发送验证码请求测试""" def test_value_stripped(self): """value 去除前后空格.""" req = SendVerificationCodeRequest(target="email", value=" u@e.com ", purpose="bind") assert req.value == "u@e.com" def test_target_purpose_preserved(self): """target 和 purpose 保持不变.""" req = SendVerificationCodeRequest(target="phone", value="13800138000", purpose="login") assert req.target == "phone" assert req.purpose == "login" # ── SendVerificationCodeResponse 测试 ────────────────────────────────────── class TestSendVerificationCodeResponse: """发送验证码响应测试""" def test_to_dict_structure(self): """to_dict 返回正确结构.""" resp = SendVerificationCodeResponse(expires_in=300, resend_after=60) d = resp.to_dict() assert d["expires_in"] == 300 assert d["resend_after"] == 60 # ── SendVerificationCodeUseCase 测试 ─────────────────────────────────────── class TestSendVerificationCode: """发送验证码用例测试""" def test_send_phone_code_success(self, mock_verification_service, mock_sms_service): """成功发送手机验证码.""" from packages.domain.verification_code import VerificationCode code_obj = VerificationCode.create("13800138000", "phone_bind", ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, sms_service=mock_sms_service, ) req = SendVerificationCodeRequest(target="phone", value="13800138000", purpose="bind") resp, error = uc.execute(req) assert error is None assert resp is not None assert resp.expires_in == 300 assert resp.resend_after == 60 mock_verification_service.generate.assert_called_once_with("13800138000", "phone_bind") mock_sms_service.send_verification_code.assert_called_once() def test_send_email_code_success(self, mock_verification_service, mock_email_service): """成功发送邮箱验证码.""" code_obj = VerificationCode.create("u@e.com", "email_bind", ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, email_service=mock_email_service, ) req = SendVerificationCodeRequest(target="email", value="U@E.COM", purpose="bind") resp, error = uc.execute(req) assert error is None # 邮箱被小写化 mock_verification_service.generate.assert_called_once_with("u@e.com", "email_bind") mock_email_service.send_email.assert_called_once() # 邮件内容包含验证码 call_args = mock_email_service.send_email.call_args[0] assert code_obj.code in call_args[2] # body def test_send_invalid_target(self, mock_verification_service): """不支持的目标类型.""" uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, ) req = SendVerificationCodeRequest(target="wechat", value="xxx", purpose="bind") resp, error = uc.execute(req) assert resp is None assert "不支持的目标类型" in error def test_send_invalid_phone_format(self, mock_verification_service): """手机号格式错误.""" uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, ) req = SendVerificationCodeRequest(target="phone", value="123", purpose="bind") resp, error = uc.execute(req) assert resp is None assert "格式" in error or "不正确" in error def test_send_invalid_email_format(self, mock_verification_service): """邮箱格式错误.""" uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, ) req = SendVerificationCodeRequest(target="email", value="invalid", purpose="bind") resp, error = uc.execute(req) assert resp is None assert "格式" in error or "不正确" in error def test_send_code_generation_fails(self, mock_verification_service): """验证码生成失败（如频控）.""" mock_verification_service.generate.return_value = (None, "发送太频繁") uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, ) req = SendVerificationCodeRequest(target="email", value="u@e.com", purpose="bind") resp, error = uc.execute(req) assert resp is None assert "发送太频繁" in error def test_send_phone_normalizes_number(self, mock_verification_service, mock_sms_service): """手机号发送时被标准化.""" code_obj = VerificationCode.create("13800138000", "phone_bind", ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, sms_service=mock_sms_service, ) req = SendVerificationCodeRequest(target="phone", value="+8613800138000", purpose="bind") uc.execute(req) # generate 收到的是标准化后的号码 call_args = mock_verification_service.generate.call_args[0] assert call_args[0] == "13800138000" def test_send_without_sms_service_still_generates(self, mock_verification_service): """没有短信服务时仍然生成验证码（但不发送）.""" code_obj = VerificationCode.create("13800138000", "phone_bind", ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, sms_service=None, ) req = SendVerificationCodeRequest(target="phone", value="13800138000", purpose="bind") resp, error = uc.execute(req) assert error is None assert resp is not None def test_send_different_purposes(self, mock_verification_service, mock_email_service): """不同 purpose 对应不同的 code_type.""" code_obj = VerificationCode.create("u@e.com", "email_login", ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, email_service=mock_email_service, ) for purpose, expected_type in [ ("bind", "email_bind"), ("login", "email_login"), ("reset_password", "email_reset_password"), ]: mock_verification_service.reset_mock() code_obj = VerificationCode.create("u@e.com", expected_type, ttl_seconds=300) mock_verification_service.generate.return_value = (code_obj, None) req = SendVerificationCodeRequest(target="email", value="u@e.com", purpose=purpose) resp, _ = uc.execute(req) assert resp is not None call_args = mock_verification_service.generate.call_args[0] assert call_args[1] == expected_type def test_send_exception_returns_error(self, mock_verification_service): """异常情况返回错误.""" mock_verification_service.generate.side_effect = Exception("Service down") uc = SendVerificationCodeUseCase( verification_code_service=mock_verification_service, ) req = SendVerificationCodeRequest(target="email", value="u@e.com", purpose="bind") resp, error = uc.execute(req) assert resp is None assert "失败" in error