灵应
23ef50ccc0
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 1h0m49s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 1h0m49s
583 lines
21 KiB
Python
583 lines
21 KiB
Python
"""
|
||
订阅支付回调单元测试
|
||
|
||
覆盖场景：
|
||
- 正确签名的回调处理（当前实现无签名验证，验证参数合法性）
|
||
- 缺失参数的回调被拒绝（422）
|
||
- 重复回调的幂等性（mark_paid 对已支付账单返回 False）
|
||
- 各种支付状态（成功处理流程）
|
||
- 不同套餐和计费周期
|
||
|
||
注：当前支付回调实现较简单（无签名验证，使用查询参数），
|
||
测试聚焦于回调处理的核心逻辑和边界情况。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
|
||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
|
||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||
|
||
from app.api.routes.subscription import _get_plan_name, _get_plan_price, router
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. Mock Billing Repository
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class MockBillingRecord:
|
||
id: str = ""
|
||
user_id: str = ""
|
||
plan_name: str = ""
|
||
amount: float = 0.0
|
||
billing_cycle: str = ""
|
||
status: str = "pending"
|
||
payment_method: str = ""
|
||
payment_id: str = ""
|
||
paid_at: datetime | None = None
|
||
created_at: datetime | None = None
|
||
|
||
def __init__(self, **kwargs):
|
||
for k, v in kwargs.items():
|
||
setattr(self, k, v)
|
||
if self.created_at is None:
|
||
self.created_at = datetime.now(timezone.utc)
|
||
|
||
|
||
class MockBillingRepository:
|
||
"""模拟的 Billing Repository，用于单元测试。"""
|
||
|
||
def __init__(self):
|
||
self.records: dict[str, MockBillingRecord] = {}
|
||
self.created_count = 0
|
||
self.mark_paid_count = 0
|
||
self.update_subscription_count = 0
|
||
self.updated_subscriptions: dict[str, dict] = {}
|
||
|
||
def create(self, record: dict) -> MockBillingRecord:
|
||
model = MockBillingRecord(**record)
|
||
self.records[model.id] = model
|
||
self.created_count += 1
|
||
return model
|
||
|
||
def find_by_user(self, user_id: str, limit: int = 50) -> list[MockBillingRecord]:
|
||
items = [r for r in self.records.values() if r.user_id == user_id]
|
||
items.sort(key=lambda r: r.created_at or datetime.min, reverse=True)
|
||
return items[:limit]
|
||
|
||
def find_by_id(self, record_id: str) -> MockBillingRecord | None:
|
||
return self.records.get(record_id)
|
||
|
||
def mark_paid(self, record_id: str, payment_method: str, payment_id: str) -> bool:
|
||
self.mark_paid_count += 1
|
||
model = self.records.get(record_id)
|
||
if model is None or model.status == "paid":
|
||
return False
|
||
model.status = "paid"
|
||
model.payment_method = payment_method
|
||
model.payment_id = payment_id
|
||
model.paid_at = datetime.now(timezone.utc)
|
||
return True
|
||
|
||
def update_subscription_on_payment(self, user_id: str, plan: str, expires_at: datetime) -> None:
|
||
self.update_subscription_count += 1
|
||
self.updated_subscriptions[user_id] = {
|
||
"plan": plan,
|
||
"expires_at": expires_at,
|
||
"status": "active",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_repo():
|
||
return MockBillingRepository()
|
||
|
||
|
||
def _make_client(mock_billing_repo: MockBillingRepository) -> TestClient:
|
||
"""创建带有 mock billing repository 的 TestClient。"""
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
|
||
# Mock SessionLocal 和 BillingRepository
|
||
mock_session = MagicMock()
|
||
with patch(
|
||
"packages.adapters.sqlalchemy_impl.session.SessionLocal",
|
||
return_value=mock_session,
|
||
):
|
||
with patch(
|
||
"packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository",
|
||
return_value=mock_billing_repo,
|
||
):
|
||
yield TestClient(test_app)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. 支付成功回调测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPaymentCallbackSuccess:
|
||
"""支付成功回调测试。"""
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_monthly_pro_payment_success(self, MockSession, MockRepo):
|
||
"""Pro 套餐月付支付成功。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={
|
||
"user_id": "user-001",
|
||
"plan": "pro",
|
||
"billing_cycle": "monthly",
|
||
"amount": 299.0,
|
||
"payment_method": "alipay",
|
||
"payment_id": "pay_20240101_001",
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["success"] is True
|
||
assert "支付成功" in data["message"]
|
||
assert "record_id" in data
|
||
|
||
# 验证账单创建
|
||
assert mock_repo.created_count == 1
|
||
# 验证标记支付
|
||
assert mock_repo.mark_paid_count == 1
|
||
# 验证订阅更新
|
||
assert mock_repo.update_subscription_count == 1
|
||
assert "user-001" in mock_repo.updated_subscriptions
|
||
assert mock_repo.updated_subscriptions["user-001"]["plan"] == "pro"
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_yearly_standard_payment_success(self, MockSession, MockRepo):
|
||
"""标准版年付支付成功。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={
|
||
"user_id": "user-002",
|
||
"plan": "standard",
|
||
"billing_cycle": "yearly",
|
||
"amount": 999.0,
|
||
"payment_method": "wechat",
|
||
"payment_id": "wx_20240101_002",
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["success"] is True
|
||
assert mock_repo.updated_subscriptions["user-002"]["plan"] == "standard"
|
||
# 年付到期时间应为约 365 天后
|
||
expires_at = mock_repo.updated_subscriptions["user-002"]["expires_at"]
|
||
expected = datetime.now(timezone.utc) + timedelta(days=365)
|
||
assert abs((expires_at - expected).days) <= 1
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_enterprise_payment_success(self, MockSession, MockRepo):
|
||
"""企业版支付成功。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={
|
||
"user_id": "user-003",
|
||
"plan": "enterprise",
|
||
"billing_cycle": "monthly",
|
||
"amount": 999.0,
|
||
"payment_method": "bank_transfer",
|
||
"payment_id": "ent_20240101_003",
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["success"] is True
|
||
assert mock_repo.updated_subscriptions["user-003"]["plan"] == "enterprise"
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_default_payment_params(self, MockSession, MockRepo):
|
||
"""使用默认 payment_method 和空 payment_id。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={
|
||
"user_id": "user-004",
|
||
"plan": "standard",
|
||
"billing_cycle": "monthly",
|
||
"amount": 99.0,
|
||
},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["success"] is True
|
||
# 默认 payment_method 应为 alipay
|
||
assert mock_repo.mark_paid_count == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. 重复回调幂等性测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPaymentCallbackIdempotency:
|
||
"""支付回调幂等性测试。"""
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_duplicate_callback_creates_new_record(self, MockSession, MockRepo):
|
||
"""重复回调（当前实现每次创建新账单，无幂等保护）。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
params = {
|
||
"user_id": "user-idem-1",
|
||
"plan": "pro",
|
||
"billing_cycle": "monthly",
|
||
"amount": 299.0,
|
||
"payment_id": "pay_dup_001",
|
||
}
|
||
|
||
# 第一次回调
|
||
resp1 = client.post("/subscription/payment-callback", params=params)
|
||
assert resp1.status_code == 200
|
||
|
||
# 第二次回调（当前实现会创建新账单，不做幂等）
|
||
resp2 = client.post("/subscription/payment-callback", params=params)
|
||
assert resp2.status_code == 200
|
||
# 当前实现每次都会创建新账单
|
||
assert mock_repo.created_count == 2
|
||
|
||
def test_mark_paid_is_idempotent(self):
|
||
"""mark_paid 方法对已支付账单返回 False（幂等）。"""
|
||
repo = MockBillingRepository()
|
||
|
||
repo.create(
|
||
dict(
|
||
id="bill-001",
|
||
user_id="u1",
|
||
plan_name="Pro 专业版",
|
||
amount=299.0,
|
||
billing_cycle="monthly",
|
||
status="pending",
|
||
)
|
||
)
|
||
|
||
# 第一次标记为已支付
|
||
result1 = repo.mark_paid("bill-001", "alipay", "pay-001")
|
||
assert result1 is True
|
||
assert repo.records["bill-001"].status == "paid"
|
||
|
||
# 第二次标记（幂等，应返回 False）
|
||
result2 = repo.mark_paid("bill-001", "alipay", "pay-001")
|
||
assert result2 is False
|
||
assert repo.records["bill-001"].status == "paid"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. 参数校验测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPaymentCallbackValidation:
|
||
"""支付回调参数校验测试。"""
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_missing_user_id_returns_422(self, MockSession, MockRepo):
|
||
"""缺少 user_id 参数返回 422。"""
|
||
MockRepo.return_value = MockBillingRepository()
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={"plan": "pro", "billing_cycle": "monthly", "amount": 299.0},
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_missing_plan_returns_422(self, MockSession, MockRepo):
|
||
"""缺少 plan 参数返回 422。"""
|
||
MockRepo.return_value = MockBillingRepository()
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={"user_id": "u1", "billing_cycle": "monthly", "amount": 299.0},
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_missing_amount_returns_422(self, MockSession, MockRepo):
|
||
"""缺少 amount 参数返回 422。"""
|
||
MockRepo.return_value = MockBillingRepository()
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={"user_id": "u1", "plan": "pro", "billing_cycle": "monthly"},
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
|
||
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
|
||
def test_negative_amount(self, MockSession, MockRepo):
|
||
"""负数金额（当前实现不校验，记录此行为）。"""
|
||
mock_repo = MockBillingRepository()
|
||
MockRepo.return_value = mock_repo
|
||
MockSession.return_value = MagicMock()
|
||
|
||
test_app = FastAPI()
|
||
test_app.include_router(router, prefix="/subscription")
|
||
client = TestClient(test_app)
|
||
|
||
resp = client.post(
|
||
"/subscription/payment-callback",
|
||
params={
|
||
"user_id": "u1",
|
||
"plan": "pro",
|
||
"billing_cycle": "monthly",
|
||
"amount": -100.0,
|
||
},
|
||
)
|
||
# 当前实现未校验金额正负
|
||
assert resp.status_code in (200, 400, 500)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. 辅助函数测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestHelperFunctions:
|
||
"""订阅辅助函数测试。"""
|
||
|
||
def test_get_plan_name_all_plans(self):
|
||
"""所有套餐名称映射正确。"""
|
||
assert _get_plan_name("free") == "体验版"
|
||
assert _get_plan_name("standard") == "标准版"
|
||
assert _get_plan_name("pro") == "专业版"
|
||
assert _get_plan_name("enterprise") == "企业版"
|
||
|
||
def test_get_plan_name_unknown(self):
|
||
"""未知套餐返回「未知套餐」。"""
|
||
assert _get_plan_name("unknown") == "未知套餐"
|
||
assert _get_plan_name("") == "未知套餐"
|
||
|
||
def test_get_plan_price_all_combinations(self):
|
||
"""所有套餐价格映射正确。"""
|
||
assert _get_plan_price("free", "monthly") == 0
|
||
assert _get_plan_price("free", "yearly") == 0
|
||
assert _get_plan_price("standard", "monthly") == 99
|
||
assert _get_plan_price("standard", "yearly") == 999
|
||
assert _get_plan_price("pro", "monthly") == 299
|
||
assert _get_plan_price("pro", "yearly") == 2999
|
||
assert _get_plan_price("enterprise", "monthly") == 999
|
||
assert _get_plan_price("enterprise", "yearly") == 9999
|
||
|
||
def test_get_plan_price_unknown(self):
|
||
"""未知组合返回 0。"""
|
||
assert _get_plan_price("unknown", "monthly") == 0
|
||
assert _get_plan_price("pro", "weekly") == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. Mock Billing Repository 单元测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMockBillingRepository:
|
||
"""Billing Repository 行为单元测试。"""
|
||
|
||
def test_create_record(self):
|
||
"""创建账单记录。"""
|
||
repo = MockBillingRepository()
|
||
record = repo.create(
|
||
dict(
|
||
id="bill-001",
|
||
user_id="user-001",
|
||
plan_name="Pro 专业版",
|
||
amount=299.0,
|
||
billing_cycle="monthly",
|
||
status="pending",
|
||
)
|
||
)
|
||
assert record.id == "bill-001"
|
||
assert record.status == "pending"
|
||
assert repo.created_count == 1
|
||
|
||
def test_find_by_id(self):
|
||
"""按 ID 查询账单。"""
|
||
repo = MockBillingRepository()
|
||
repo.create(
|
||
dict(
|
||
id="bill-001",
|
||
user_id="user-1",
|
||
plan_name="Pro",
|
||
amount=299,
|
||
billing_cycle="monthly",
|
||
status="pending",
|
||
)
|
||
)
|
||
|
||
found = repo.find_by_id("bill-001")
|
||
assert found is not None
|
||
assert found.id == "bill-001"
|
||
|
||
not_found = repo.find_by_id("nonexistent")
|
||
assert not_found is None
|
||
|
||
def test_find_by_user(self):
|
||
"""按用户查询账单。"""
|
||
repo = MockBillingRepository()
|
||
repo.create(dict(id="b1", user_id="u1", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
|
||
repo.create(
|
||
dict(id="b2", user_id="u1", plan_name="Standard", amount=99, billing_cycle="monthly", status="pending")
|
||
)
|
||
repo.create(dict(id="b3", user_id="u2", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
|
||
|
||
user1_records = repo.find_by_user("u1")
|
||
assert len(user1_records) == 2
|
||
|
||
user2_records = repo.find_by_user("u2")
|
||
assert len(user2_records) == 1
|
||
|
||
def test_mark_paid_transitions_status(self):
|
||
"""mark_paid 正确转换状态。"""
|
||
repo = MockBillingRepository()
|
||
repo.create(
|
||
dict(
|
||
id="bill-001",
|
||
user_id="u1",
|
||
plan_name="Pro",
|
||
amount=299,
|
||
billing_cycle="monthly",
|
||
status="pending",
|
||
)
|
||
)
|
||
|
||
result = repo.mark_paid("bill-001", "alipay", "pay-001")
|
||
assert result is True
|
||
|
||
record = repo.find_by_id("bill-001")
|
||
assert record.status == "paid"
|
||
assert record.payment_method == "alipay"
|
||
assert record.payment_id == "pay-001"
|
||
assert record.paid_at is not None
|
||
|
||
def test_mark_paid_idempotent(self):
|
||
"""mark_paid 对已支付账单幂等。"""
|
||
repo = MockBillingRepository()
|
||
repo.create(
|
||
dict(
|
||
id="bill-001",
|
||
user_id="u1",
|
||
plan_name="Pro",
|
||
amount=299,
|
||
billing_cycle="monthly",
|
||
status="pending",
|
||
)
|
||
)
|
||
|
||
repo.mark_paid("bill-001", "alipay", "pay-001")
|
||
paid_at_first = repo.find_by_id("bill-001").paid_at
|
||
|
||
result = repo.mark_paid("bill-001", "alipay", "pay-001")
|
||
assert result is False
|
||
# paid_at 不应更新
|
||
assert repo.find_by_id("bill-001").paid_at == paid_at_first
|
||
|
||
def test_mark_paid_nonexistent_returns_false(self):
|
||
"""标记不存在的账单返回 False。"""
|
||
repo = MockBillingRepository()
|
||
result = repo.mark_paid("nonexistent", "alipay", "pay-001")
|
||
assert result is False
|
||
|
||
def test_update_subscription_on_payment(self):
|
||
"""支付成功后更新订阅。"""
|
||
repo = MockBillingRepository()
|
||
expires = datetime.now(timezone.utc) + timedelta(days=30)
|
||
|
||
repo.update_subscription_on_payment("user-001", "pro", expires)
|
||
|
||
assert repo.update_subscription_count == 1
|
||
assert "user-001" in repo.updated_subscriptions
|
||
sub = repo.updated_subscriptions["user-001"]
|
||
assert sub["plan"] == "pro"
|
||
assert sub["status"] == "active"
|
||
assert sub["expires_at"] == expires
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|