xiaoxia-saas/tests/unit/test_auth_bind_wechat_sync.py

"""Auth bind_contact + wechat_sync use cases unit tests.

Covers BindContactUseCase, SendVerificationCodeUseCase, WechatSyncUseCase.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock

from packages.application.auth.bind_contact_use_case import (
    BindContactRequest,
    BindContactResponse,
    BindContactUseCase,
    SendVerificationCodeRequest,
    SendVerificationCodeResponse,
    SendVerificationCodeUseCase,
)
from packages.application.auth.wechat_sync_use_case import (
    WechatSyncRequest,
    WechatSyncResponse,
    WechatSyncUseCase,
)

# ── Test helpers ─────────────────────────────────────────


@dataclass
class FakeUser:
    id: str = "user-123"
    email: str = "test@example.com"
    display_name: str = "Test User"
    username: str = "testuser"
    password_hash: str = ""
    email_verified: bool = False
    phone: str = ""
    phone_verified: bool = False
    binding_completed_at: datetime | None = None
    last_login_at: datetime | None = None
    last_login_ip: str | None = None
    wechat_openid: str | None = None
    wechat_unionid: str | None = None
    email_verification_token: str | None = None
    password_reset_token: str | None = None
    password_reset_expires_at: datetime | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


class FakeUserRepository:
    def __init__(self, user=None):
        self._user = user
        self.saved_user = None
        self.save_called = 0

    def find_by_id(self, user_id):
        if self._user and self._user.id == user_id:
            return self._user
        return None

    def find_by_email(self, email):
        if self._user and self._user.email == email:
            return self._user
        return None

    def find_by_phone(self, phone):
        if self._user and self._user.phone == phone:
            return self._user
        return None

    def find_by_username(self, username):
        if self._user and self._user.username == username:
            return self._user
        return None

    def find_by_wechat_openid(self, openid):
        if self._user and self._user.wechat_openid == openid:
            return self._user
        return None

    def find_by_wechat_unionid(self, unionid):
        if self._user and self._user.wechat_unionid == unionid:
            return self._user
        return None

    def save(self, user):
        self.saved_user = user
        self.save_called += 1
        self._user = user
        return user


class FakeVerificationCode:
    def __init__(self, code="123456", created_at=None, expires_at=None):
        self.code = code
        self.created_at = created_at or datetime.now(timezone.utc)
        self.expires_at = expires_at or (datetime.now(timezone.utc) + timedelta(minutes=5))


class FakeVerificationCodeService:
    def __init__(self, verify_success=True, verify_error=None, generate_code="123456"):
        self._verify_success = verify_success
        self._verify_error = verify_error
        self._generate_code = generate_code
        self.verified = []
        self.generated = []

    def verify(self, recipient, code_type, code_value):
        self.verified.append(
            {
                "recipient": recipient,
                "code_type": code_type,
                "code_value": code_value,
            }
        )
        return self._verify_success, self._verify_error

    def generate(self, recipient, code_type):
        self.generated.append({"recipient": recipient, "code_type": code_type})
        return FakeVerificationCode(code=self._generate_code), None


class FakeSessionStore:
    def __init__(self):
        self.saved_sessions = []

    def save_session(self, **kwargs):
        self.saved_sessions.append(kwargs)
        return True


class FakeSmsService:
    def __init__(self):
        self.sent = []

    def send_verification_code(self, phone, code):
        self.sent.append({"phone": phone, "code": code})


class FakeEmailService:
    def __init__(self):
        self.sent = []

    def send_email(self, to, subject, body):
        self.sent.append({"to": to, "subject": subject, "body": body})


# ── BindContactUseCase tests ────────────────────────────


class TestBindContactUseCase:
    def _make_use_case(self, user_repo=None, verify_svc=None):
        return BindContactUseCase(
            user_repository=user_repo or FakeUserRepository(),
            verification_code_service=verify_svc or FakeVerificationCodeService(),
        )

    def test_bind_phone_success(self):
        user = FakeUser(id="user-1", phone="", phone_verified=False)
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(
            user_id="user-1",
            phone="13800138000",
            phone_code="123456",
        )
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert isinstance(response, BindContactResponse)
        assert response.user.phone == "13800138000"
        assert response.user.phone_verified is True

        # verification was called
        assert len(verify_svc.verified) == 1
        assert verify_svc.verified[0]["recipient"] == "13800138000"

    def test_bind_email_success(self):
        user = FakeUser(id="user-1", email="old@example.com", email_verified=False)
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(
            user_id="user-1",
            email="new@example.com",
            email_code="123456",
        )
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert response.user.email == "new@example.com"
        assert response.user.email_verified is True
        assert len(verify_svc.verified) == 1

    def test_bind_both_phone_and_email(self):
        user = FakeUser(id="user-1", phone="", phone_verified=False, email_verified=False)
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(
            user_id="user-1",
            phone="13800138000",
            phone_code="123456",
            email="new@example.com",
            email_code="123456",
        )
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert response.user.phone == "13800138000"
        assert response.user.phone_verified is True
        assert response.user.email == "new@example.com"
        assert response.user.email_verified is True
        assert response.user.binding_completed_at is not None
        assert len(verify_svc.verified) == 2

    def test_binding_complete_with_real_email(self):
        """Both phone and email verified, real email (not wechat.local) → binding completed."""
        user = FakeUser(
            id="user-1",
            email="",
            phone="",
            phone_verified=False,
            email_verified=False,
        )
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(
            user_id="user-1",
            phone="13800138000",
            phone_code="123456",
            email="user@real.com",
            email_code="123456",
        )
        response, error = use_case.execute(req)

        assert error is None
        assert response.user.binding_completed_at is not None
        assert response.to_dict()["user"]["binding_complete"] is True

    def test_binding_not_complete_with_wechat_email(self):
        """WeChat placeholder email doesn't count for binding completion."""
        user = FakeUser(
            id="user-1",
            email="abc@wechat.local",
            email_verified=True,
            phone="",
            phone_verified=False,
        )
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(
            user_id="user-1",
            phone="13800138000",
            phone_code="123456",
        )
        response, error = use_case.execute(req)

        assert error is None
        # wechat.local email doesn't count
        assert response.user.binding_completed_at is None
        assert response.to_dict()["user"]["binding_complete"] is False

    def test_no_phone_no_email_returns_error(self):
        repo = FakeUserRepository()
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="user-1")
        response, error = use_case.execute(req)

        assert response is None
        assert "至少填写手机号或邮箱" in error

    def test_user_not_found(self):
        repo = FakeUserRepository()  # no user
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="nonexistent", phone="13800138000", phone_code="123")
        response, error = use_case.execute(req)

        assert response is None
        assert "用户不存在" in error

    def test_invalid_phone_format(self):
        user = FakeUser(id="user-1")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="user-1", phone="123", phone_code="123456")
        response, error = use_case.execute(req)

        assert response is None
        assert error is not None

    def test_invalid_email_format(self):
        user = FakeUser(id="user-1")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="user-1", email="not-an-email", email_code="123456")
        response, error = use_case.execute(req)

        assert response is None
        assert error is not None

    def test_phone_missing_code(self):
        user = FakeUser(id="user-1")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="user-1", phone="13800138000")
        response, error = use_case.execute(req)

        assert response is None
        assert "请输入手机验证码" in error

    def test_email_missing_code(self):
        user = FakeUser(id="user-1")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = BindContactRequest(user_id="user-1", email="a@b.com")
        response, error = use_case.execute(req)

        assert response is None
        assert "请输入邮箱验证码" in error

    def test_phone_already_bound_to_other_user(self):
        other_user = FakeUser(id="user-2", phone="13800138000")
        current_user = FakeUser(id="user-1", phone="")
        # repo only finds "other" user for phone lookup
        repo = FakeUserRepository(user=other_user)
        # But we also need find_by_id to find the current user
        # Our simple repo can only hold one user. Let's use MagicMock instead.
        repo = MagicMock()
        repo.find_by_id.return_value = current_user
        repo.find_by_phone.return_value = other_user

        verify_svc = FakeVerificationCodeService()
        use_case = BindContactUseCase(user_repository=repo, verification_code_service=verify_svc)

        req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456")
        response, error = use_case.execute(req)

        assert response is None
        assert "已被其他账号绑定" in error

    def test_email_already_bound_to_other_user(self):
        current_user = FakeUser(id="user-1", email="old@example.com")
        other_user = FakeUser(id="user-2", email="new@example.com")

        repo = MagicMock()
        repo.find_by_id.return_value = current_user
        repo.find_by_email.return_value = other_user

        verify_svc = FakeVerificationCodeService()
        use_case = BindContactUseCase(user_repository=repo, verification_code_service=verify_svc)

        req = BindContactRequest(user_id="user-1", email="new@example.com", email_code="123456")
        response, error = use_case.execute(req)

        assert response is None
        assert "已被其他账号绑定" in error

    def test_phone_verification_failed(self):
        user = FakeUser(id="user-1", phone="")
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=False, verify_error="验证码错误或已过期")
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="wrong")
        response, error = use_case.execute(req)

        assert response is None
        assert "手机验证码错误" in error

    def test_email_verification_failed(self):
        user = FakeUser(id="user-1")
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=False, verify_error="验证码错误")
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(user_id="user-1", email="a@b.com", email_code="wrong")
        response, error = use_case.execute(req)

        assert response is None
        assert "邮箱验证码错误" in error

    def test_bind_same_phone_to_self_ok(self):
        """Binding the same phone to the same user should work (no conflict)."""
        user = FakeUser(id="user-1", phone="13800138000", phone_verified=False)
        repo = FakeUserRepository(user=user)
        verify_svc = FakeVerificationCodeService(verify_success=True)
        use_case = self._make_use_case(user_repo=repo, verify_svc=verify_svc)

        req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123456")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None

    def test_exception_handling(self):
        repo = MagicMock()
        repo.find_by_id.side_effect = RuntimeError("DB down")
        verify_svc = FakeVerificationCodeService()
        use_case = BindContactUseCase(user_repository=repo, verification_code_service=verify_svc)

        req = BindContactRequest(user_id="user-1", phone="13800138000", phone_code="123")
        response, error = use_case.execute(req)

        assert response is None
        assert "绑定失败" in error


# ── SendVerificationCodeUseCase tests ───────────────────


class TestSendVerificationCodeUseCase:
    def test_send_phone_code_success(self):
        verify_svc = FakeVerificationCodeService(generate_code="654321")
        sms_svc = FakeSmsService()
        use_case = SendVerificationCodeUseCase(
            verification_code_service=verify_svc,
            sms_service=sms_svc,
        )

        req = SendVerificationCodeRequest(target="phone", value="13800138000", purpose="bind")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert isinstance(response, SendVerificationCodeResponse)
        assert response.expires_in > 0
        assert response.resend_after == 60

        assert len(verify_svc.generated) == 1
        assert verify_svc.generated[0]["code_type"] == "phone_bind"

        # SMS was sent
        assert len(sms_svc.sent) == 1
        assert sms_svc.sent[0]["phone"] == "13800138000"
        assert sms_svc.sent[0]["code"] == "654321"

    def test_send_email_code_success(self):
        verify_svc = FakeVerificationCodeService(generate_code="111222")
        email_svc = FakeEmailService()
        use_case = SendVerificationCodeUseCase(
            verification_code_service=verify_svc,
            email_service=email_svc,
        )

        req = SendVerificationCodeRequest(target="email", value="test@example.com", purpose="bind")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None

        assert len(verify_svc.generated) == 1
        assert verify_svc.generated[0]["code_type"] == "email_bind"

        assert len(email_svc.sent) == 1
        assert email_svc.sent[0]["to"] == "test@example.com"
        assert "111222" in email_svc.sent[0]["body"]

    def test_invalid_phone_format(self):
        verify_svc = FakeVerificationCodeService()
        use_case = SendVerificationCodeUseCase(verification_code_service=verify_svc)

        req = SendVerificationCodeRequest(target="phone", value="123", purpose="bind")
        response, error = use_case.execute(req)

        assert response is None
        assert error is not None

    def test_invalid_email_format(self):
        verify_svc = FakeVerificationCodeService()
        use_case = SendVerificationCodeUseCase(verification_code_service=verify_svc)

        req = SendVerificationCodeRequest(target="email", value="not-email", purpose="bind")
        response, error = use_case.execute(req)

        assert response is None
        assert error is not None

    def test_unsupported_target_type(self):
        verify_svc = FakeVerificationCodeService()
        use_case = SendVerificationCodeUseCase(verification_code_service=verify_svc)

        req = SendVerificationCodeRequest(target="carrier_pigeon", value="hello", purpose="bind")
        response, error = use_case.execute(req)

        assert response is None
        assert "不支持的目标类型" in error

    def test_email_recipient_lowercased(self):
        verify_svc = FakeVerificationCodeService()
        email_svc = FakeEmailService()
        use_case = SendVerificationCodeUseCase(
            verification_code_service=verify_svc,
            email_service=email_svc,
        )

        req = SendVerificationCodeRequest(target="email", value="TEST@Example.COM", purpose="login")
        response, error = use_case.execute(req)

        assert error is None
        # recipient should be lowercased
        assert verify_svc.generated[0]["recipient"] == "test@example.com"

    def test_no_sms_service_phone_still_returns_success(self):
        """If no SMS service is configured, code is generated but not sent."""
        verify_svc = FakeVerificationCodeService()
        use_case = SendVerificationCodeUseCase(verification_code_service=verify_svc)

        req = SendVerificationCodeRequest(target="phone", value="13800138000", purpose="login")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        # code was generated
        assert len(verify_svc.generated) == 1

    def test_exception_handling(self):
        verify_svc = MagicMock()
        verify_svc.generate.side_effect = RuntimeError("Redis down")
        use_case = SendVerificationCodeUseCase(verification_code_service=verify_svc)

        req = SendVerificationCodeRequest(target="email", value="a@b.com", purpose="bind")
        response, error = use_case.execute(req)

        assert response is None
        assert "发送失败" in error

    def test_to_dict_returns_correct_fields(self):
        resp = SendVerificationCodeResponse(expires_in=300, resend_after=60)
        d = resp.to_dict()
        assert d["expires_in"] == 300
        assert d["resend_after"] == 60


# ── WechatSyncUseCase tests ─────────────────────────────


class TestWechatSyncUseCase:
    def _make_use_case(self, user_repo=None, session_store=None, secret_key="test-secret-key-for-jwt"):
        return WechatSyncUseCase(
            user_repository=user_repo or FakeUserRepository(),
            session_store=session_store or FakeSessionStore(),
            jwt_secret_key=secret_key,
        )

    def test_existing_user_login_by_openid(self):
        user = FakeUser(
            id="user-1",
            wechat_openid="openid-abc",
            display_name="WeChat User",
        )
        repo = FakeUserRepository(user=user)
        store = FakeSessionStore()
        use_case = self._make_use_case(user_repo=repo, session_store=store)

        req = WechatSyncRequest(openid="openid-abc")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert isinstance(response, WechatSyncResponse)
        assert response.user_id == "user-1"
        assert response.is_new_user is False
        assert response.access_token
        assert response.refresh_token

        # session created
        assert len(store.saved_sessions) == 1
        session = store.saved_sessions[0]
        assert session["user_id"] == "user-1"
        assert "wechat_" in session["device_info"]

        # last login updated
        assert repo.saved_user.last_login_at is not None
        assert repo.saved_user.last_login_ip == "bff_gateway"

    def test_existing_user_by_unionid_binds_openid(self):
        user = FakeUser(
            id="user-1",
            wechat_openid=None,  # no openid
            wechat_unionid="unionid-xyz",
            display_name="Existing User",
        )
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = WechatSyncRequest(openid="openid-new", unionid="unionid-xyz")
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert response.user_id == "user-1"
        assert response.is_new_user is False

        # openid was bound
        assert repo.saved_user.wechat_openid == "openid-new"

    def test_new_user_creation(self):
        repo = FakeUserRepository()  # no existing user
        store = FakeSessionStore()
        use_case = self._make_use_case(user_repo=repo, session_store=store)

        req = WechatSyncRequest(
            openid="openid-new123",
            nickname="微信昵称",
            avatar_url="https://example.com/avatar.png",
        )
        response, error = use_case.execute(req)

        assert error is None
        assert response is not None
        assert response.is_new_user is True
        assert response.nickname == "微信昵称"
        assert response.user_id
        assert len(response.user_id) == 32  # uuid4 hex

        # user was saved
        assert repo.saved_user is not None
        assert repo.saved_user.wechat_openid == "openid-new123"
        assert repo.saved_user.email_verified is True
        assert "wechat.local" in repo.saved_user.email
        assert repo.saved_user.username.startswith("wx_")
        assert repo.saved_user.display_name == "微信昵称"

    def test_empty_openid_returns_error(self):
        use_case = self._make_use_case()
        req = WechatSyncRequest(openid="")
        response, error = use_case.execute(req)

        assert response is None
        assert "openid is required" in error

    def test_default_nickname_when_empty(self):
        repo = FakeUserRepository()
        use_case = self._make_use_case(user_repo=repo)

        req = WechatSyncRequest(openid="openid-1", nickname="")
        response, error = use_case.execute(req)

        assert error is None
        assert response.is_new_user is True
        assert response.nickname == "微信用户"

    def test_username_uniqueness_suffix(self):
        """When username already exists, a numeric suffix is added."""
        # First user with same openid prefix
        existing = FakeUser(username="wx_openidnew123_")
        repo = FakeUserRepository(user=existing)

        # Our simple FakeUserRepository only holds one user.
        # Use MagicMock for more control.
        repo = MagicMock()
        repo.find_by_wechat_openid.return_value = None
        repo.find_by_wechat_unionid.return_value = None
        # first find_by_username returns a user (conflict), second time None (unique)
        call_count = {"n": 0}

        def mock_find_by_username(username):
            call_count["n"] += 1
            if call_count["n"] == 1:
                return FakeUser(username=username)  # conflict
            return None  # unique on second try

        repo.find_by_username.side_effect = mock_find_by_username
        repo.save = MagicMock(side_effect=lambda u: u)

        use_case = WechatSyncUseCase(
            user_repository=repo,
            session_store=FakeSessionStore(),
            jwt_secret_key="test-secret",
        )

        req = WechatSyncRequest(openid="openid-new123", nickname="Test")
        response, error = use_case.execute(req)

        assert error is None
        assert response.is_new_user is True
        # username should have _1 suffix
        saved_user = repo.save.call_args[0][0]
        assert saved_user.username.endswith("_1")

    def test_to_dict_has_token_alias_for_compat(self):
        user = FakeUser(id="u-1", wechat_openid="oid-1", display_name="Name")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = WechatSyncRequest(openid="oid-1")
        response, _ = use_case.execute(req)

        d = response.to_dict()
        assert d["access_token"] == d["token"]  # compat alias
        assert d["is_new_user"] is False
        assert d["user"]["id"] == "u-1"
        assert d["user_info"]["display_name"] == "Name"

    def test_access_token_has_correct_claims(self):
        import jwt as pyjwt

        user = FakeUser(id="user-99", wechat_openid="oid-99")
        repo = FakeUserRepository(user=user)
        use_case = self._make_use_case(user_repo=repo)

        req = WechatSyncRequest(openid="oid-99")
        response, _ = use_case.execute(req)

        payload = pyjwt.decode(response.access_token, "test-secret-key-for-jwt", algorithms=["HS256"])
        assert payload["sub"] == "user-99"
        assert payload["type"] == "user_auth"
        assert "sid" in payload
        assert "exp" in payload

    def test_exception_handling(self):
        repo = MagicMock()
        repo.find_by_wechat_openid.side_effect = RuntimeError("DB down")
        use_case = self._make_use_case(user_repo=repo)

        req = WechatSyncRequest(openid="abc")
        response, error = use_case.execute(req)

        assert response is None
        assert "Internal error" in error