ef1aae71d2
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 14h2m19s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 14h2m29s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2s
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 13s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 14s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 14h1m35s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 14h1m35s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 14h1m44s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 14h1m45s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 14h14m34s
112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
"""注册送积分单元测试 (#1895 P2 step 3)"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_settings(monkeypatch):
|
|
"""默认关闭 points_enabled,不影响现有用例。"""
|
|
from app.config import settings
|
|
|
|
monkeypatch.setattr(settings, "points_enabled", False)
|
|
return settings
|
|
|
|
|
|
class TestRegisterBonusPoints:
|
|
@pytest.mark.asyncio
|
|
async def test_bonus_when_enabled(self, mock_settings):
|
|
"""开启积分时注册成功送50分。"""
|
|
from app.api.routes import auth
|
|
from app.api.routes.auth import RegisterRequest
|
|
|
|
mock_settings.points_enabled = True
|
|
|
|
mock_uc = MagicMock()
|
|
mock_resp = MagicMock()
|
|
mock_resp.user_id = "new-user-1"
|
|
mock_resp.email = "n***@example.com"
|
|
mock_resp.username = "newuser"
|
|
mock_resp.display_name = "New User"
|
|
mock_uc.execute.return_value = (mock_resp, None)
|
|
|
|
mock_svc = MagicMock()
|
|
|
|
def _mock_uc_cls(*args, **kwargs):
|
|
return mock_uc
|
|
|
|
db = MagicMock()
|
|
|
|
with (
|
|
patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
|
|
patch("packages.domain.points_service.PointsService", return_value=mock_svc),
|
|
):
|
|
req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser")
|
|
resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
|
|
assert resp.user_id == "new-user-1"
|
|
mock_svc.add_points.assert_called_once()
|
|
call_kwargs = mock_svc.add_points.call_args.kwargs
|
|
assert call_kwargs["user_id"] == "new-user-1"
|
|
assert call_kwargs["amount"] == 50
|
|
assert call_kwargs["source"] == "task_reward"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_bonus_when_disabled(self, mock_settings):
|
|
"""关闭积分时不送分。"""
|
|
from app.api.routes import auth
|
|
from app.api.routes.auth import RegisterRequest
|
|
|
|
mock_uc = MagicMock()
|
|
mock_resp = MagicMock()
|
|
mock_resp.user_id = "new-user-2"
|
|
mock_resp.email = "n***@example.com"
|
|
mock_resp.username = "newuser2"
|
|
mock_resp.display_name = "New User 2"
|
|
mock_uc.execute.return_value = (mock_resp, None)
|
|
|
|
def _mock_uc_cls(*args, **kwargs):
|
|
return mock_uc
|
|
|
|
db = MagicMock()
|
|
with (
|
|
patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
|
|
patch("packages.domain.points_service.PointsService") as MockSvc,
|
|
):
|
|
req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser2")
|
|
resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
|
|
MockSvc.assert_not_called()
|
|
assert resp.user_id == "new-user-2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bonus_failure_does_not_break_register(self, mock_settings):
|
|
"""送积分失败不应影响注册流程。"""
|
|
from app.api.routes import auth
|
|
from app.api.routes.auth import RegisterRequest
|
|
|
|
mock_settings.points_enabled = True
|
|
mock_uc = MagicMock()
|
|
mock_resp = MagicMock()
|
|
mock_resp.user_id = "new-user-3"
|
|
mock_resp.email = "n***@example.com"
|
|
mock_resp.username = "newuser3"
|
|
mock_resp.display_name = "New User 3"
|
|
mock_uc.execute.return_value = (mock_resp, None)
|
|
|
|
mock_svc = MagicMock()
|
|
mock_svc.add_points.side_effect = Exception("DB error")
|
|
|
|
def _mock_uc_cls(*args, **kwargs):
|
|
return mock_uc
|
|
|
|
db = MagicMock()
|
|
with (
|
|
patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
|
|
patch("packages.domain.points_service.PointsService", return_value=mock_svc),
|
|
):
|
|
req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser3")
|
|
resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
|
|
assert resp.user_id == "new-user-3"
|