Реализация Premium-подписки, рефакторинг, интеграция с API. Полный цикл от red до green
Часть 2: интегрируем Premium-подписку с платежами и API.
# tests/test_subscription_activation.py
from datetime import datetime, timedelta
from todo_app.services import SubscriptionService
def test_activate_monthly_subscription():
"""RED: Активация подписки"""
user = User(email="user@test.com")
service = SubscriptionService()
service.activate_subscription(
user=user,
plan=SubscriptionPlan.PREMIUM_MONTHLY
)
assert user.subscription_plan == SubscriptionPlan.PREMIUM_MONTHLY
assert user.is_premium() is True
# Expires in ~30 days
assert user.subscription_expires_at > datetime.now() + timedelta(days=29)
def test_activate_annual_subscription():
"""Annual expires in 365 days"""
user = User(email="user@test.com")
service = SubscriptionService()
service.activate_subscription(
user=user,
plan=SubscriptionPlan.PREMIUM_ANNUAL
)
# Expires in ~365 days
assert user.subscription_expires_at > datetime.now() + timedelta(days=364)# todo_app/services.py
from datetime import datetime, timedelta
class SubscriptionService:
def activate_subscription(self, user: User, plan: SubscriptionPlan):
"""Активировать подписку"""
user.subscription_plan = plan
if plan == SubscriptionPlan.PREMIUM_MONTHLY:
duration = timedelta(days=30)
elif plan == SubscriptionPlan.PREMIUM_ANNUAL:
duration = timedelta(days=365)
else:
raise ValueError(f"Cannot activate {plan}")
user.subscription_expires_at = datetime.now() + duration✅ PASSED
# tests/test_subscription_purchase.py
def test_purchase_premium_monthly():
"""RED: Покупка подписки с платежом"""
user = User(email="user@test.com")
mock_payment = Mock()
mock_payment.charge.return_value = {"status": "success", "charge_id": "ch_123"}
subscription_service = SubscriptionService(payment_gateway=mock_payment)
result = subscription_service.purchase_subscription(
user=user,
plan=SubscriptionPlan.PREMIUM_MONTHLY,
card_token="tok_visa"
)
# Проверки
assert result["status"] == "success"
assert user.is_premium() is True
# Платёж совершён
mock_payment.charge.assert_called_once_with(
amount_cents=999, # $9.99
card_token="tok_visa",
description="Premium Monthly - user@test.com"
)class SubscriptionService:
def __init__(self, payment_gateway: PaymentGateway):
self.payment_gateway = payment_gateway
def purchase_subscription(
self,
user: User,
plan: SubscriptionPlan,
card_token: str
):
"""Купить подписку"""
# Платёж
amount_cents = int(plan.get_price() * 100)
charge_result = self.payment_gateway.charge(
amount_cents=amount_cents,
card_token=card_token,
description=f"{plan.plan_id.title()} - {user.email}"
)
if charge_result["status"] != "success":
return charge_result
# Активация
self.activate_subscription(user, plan)
return {"status": "success", "subscription": plan.plan_id}✅ PASSED
# tests/test_subscription_api.py
import pytest
from fastapi.testclient import TestClient
from todo_app.api import app
@pytest.fixture
def client():
return TestClient(app)
def test_subscribe_to_premium_api(client):
"""RED: API endpoint ещё не существует"""
response = client.post(
"/api/subscriptions/subscribe",
json={
"user_email": "user@test.com",
"plan": "premium_monthly",
"card_token": "tok_visa"
}
)
assert response.status_code == 200
assert response.json()["status"] == "success"# todo_app/api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class SubscribeRequest(BaseModel):
user_email: str
plan: str
card_token: str
@app.post("/api/subscriptions/subscribe")
def subscribe_to_premium(request: SubscribeRequest):
"""API endpoint для подписки"""
# TODO: Получить user из БД
user = User(email=request.user_email)
try:
plan = SubscriptionPlan[request.plan.upper()]
except KeyError:
raise HTTPException(status_code=400, detail="Invalid plan")
# Payment gateway (DI через dependency)
payment_gateway = StripeGateway(api_key=os.getenv("STRIPE_API_KEY"))
subscription_service = SubscriptionService(payment_gateway)
result = subscription_service.purchase_subscription(
user=user,
plan=plan,
card_token=request.card_token
)
if result["status"] != "success":
raise HTTPException(status_code=402, detail=result.get("message"))
return result✅ PASSED
def test_subscription_needs_renewal():
"""RED: Проверка что подписка скоро истечёт"""
user = User(
email="user@test.com",
subscription_plan=SubscriptionPlan.PREMIUM_MONTHLY,
subscription_expires_at=datetime.now() + timedelta(days=3) # 3 дня до истечения
)
assert user.needs_renewal() is True
def test_subscription_no_renewal_needed():
"""Подписка активна ещё долго"""
user = User(
email="user@test.com",
subscription_plan=SubscriptionPlan.PREMIUM_MONTHLY,
subscription_expires_at=datetime.now() + timedelta(days=20)
)
assert user.needs_renewal() is False@dataclass
class User:
# ...
def needs_renewal(self, days_before=7) -> bool:
"""Проверка что подписка истекает скоро"""
if not self.is_premium():
return False
if self.subscription_expires_at is None:
return False
days_left = (self.subscription_expires_at - datetime.now()).days
return days_left <= days_before✅ PASSED
# todo_app/policies.py
from dataclasses import dataclass
from datetime import timedelta
@dataclass
class SubscriptionPolicy:
"""Бизнес-правила подписок"""
FREE_TODO_LIMIT = 10
RENEWAL_REMINDER_DAYS = 7
@staticmethod
def get_duration(plan: SubscriptionPlan) -> timedelta:
"""Длительность подписки"""
durations = {
SubscriptionPlan.PREMIUM_MONTHLY: timedelta(days=30),
SubscriptionPlan.PREMIUM_ANNUAL: timedelta(days=365),
}
return durations[plan]
@staticmethod
def get_discount(plan: SubscriptionPlan) -> float:
"""Скидка для плана"""
if plan == SubscriptionPlan.PREMIUM_ANNUAL:
return 0.17 # 17% off
return 0.0Обновить SubscriptionService:
class SubscriptionService:
def activate_subscription(self, user: User, plan: SubscriptionPlan):
"""Активировать с использованием policy"""
user.subscription_plan = plan
duration = SubscriptionPolicy.get_duration(plan)
user.subscription_expires_at = datetime.now() + duration✅ Тесты всё ещё проходят!
# tests/test_subscription_e2e.py
def test_full_subscription_flow():
"""E2E: От покупки до истечения"""
# 1. Free user создаёт 10 задач
user = User(email="user@test.com")
service = TodoService(mock_repo)
for i in range(10):
service.create_todo(user, f"Task {i+1}", Priority.MEDIUM)
# 2. Пытается создать 11-ю (должно упасть)
with pytest.raises(TodoLimitExceeded):
service.create_todo(user, "Task 11", Priority.HIGH)
# 3. Покупает Premium
subscription_service = SubscriptionService(mock_payment)
subscription_service.purchase_subscription(
user, SubscriptionPlan.PREMIUM_MONTHLY, "tok_visa"
)
# 4. Теперь может создавать unlimited
for i in range(100):
service.create_todo(user, f"Premium Task {i+1}", Priority.LOW)
# 5. Подписка истекает
user.subscription_expires_at = datetime.now() - timedelta(days=1)
# 6. Снова лимит 10 задач
assert user.is_premium() is FalseВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: TDD: Когда работает, когда мешает