|
import pytest
|
|
import pytest_asyncio
|
|
from decimal import Decimal
|
|
from datetime import date, datetime
|
|
from tortoise.contrib.test import initializer, finalizer
|
|
from rbox.models import User
|
|
from rbox.billing.models import (
|
|
SubscriptionPlan, UserSubscription, UsageRecord, UsageAggregate,
|
|
Invoice, InvoiceLineItem, PricingConfig, PaymentMethod, BillingEvent
|
|
)
|
|
|
|
@pytest.fixture(scope="module")
|
|
def event_loop():
|
|
import asyncio
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
@pytest_asyncio.fixture(scope="module", autouse=True)
|
|
async def initialize_tests():
|
|
initializer(["rbox.models", "rbox.billing.models"], db_url="sqlite://:memory:")
|
|
yield
|
|
await finalizer()
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_user():
|
|
user = await User.create(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
hashed_password="hashed_password_here",
|
|
is_active=True
|
|
)
|
|
yield user
|
|
await user.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscription_plan_creation():
|
|
plan = await SubscriptionPlan.create(
|
|
name="starter",
|
|
display_name="Starter Plan",
|
|
description="Basic storage plan",
|
|
storage_gb=100,
|
|
bandwidth_gb=100,
|
|
price_monthly=Decimal("5.00"),
|
|
price_yearly=Decimal("50.00")
|
|
)
|
|
|
|
assert plan.name == "starter"
|
|
assert plan.storage_gb == 100
|
|
assert plan.price_monthly == Decimal("5.00")
|
|
await plan.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_subscription_creation(test_user):
|
|
subscription = await UserSubscription.create(
|
|
user=test_user,
|
|
billing_type="pay_as_you_go",
|
|
status="active"
|
|
)
|
|
|
|
assert subscription.user_id == test_user.id
|
|
assert subscription.billing_type == "pay_as_you_go"
|
|
assert subscription.status == "active"
|
|
await subscription.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_usage_record_creation(test_user):
|
|
usage = await UsageRecord.create(
|
|
user=test_user,
|
|
record_type="storage",
|
|
amount_bytes=1024 * 1024 * 100,
|
|
resource_type="file",
|
|
resource_id=1,
|
|
idempotency_key="test_key_123"
|
|
)
|
|
|
|
assert usage.user_id == test_user.id
|
|
assert usage.record_type == "storage"
|
|
assert usage.amount_bytes == 1024 * 1024 * 100
|
|
await usage.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_usage_aggregate_creation(test_user):
|
|
aggregate = await UsageAggregate.create(
|
|
user=test_user,
|
|
date=date.today(),
|
|
storage_bytes_avg=1024 * 1024 * 500,
|
|
storage_bytes_peak=1024 * 1024 * 600,
|
|
bandwidth_up_bytes=1024 * 1024 * 50,
|
|
bandwidth_down_bytes=1024 * 1024 * 100
|
|
)
|
|
|
|
assert aggregate.user_id == test_user.id
|
|
assert aggregate.storage_bytes_avg == 1024 * 1024 * 500
|
|
await aggregate.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invoice_creation(test_user):
|
|
invoice = await Invoice.create(
|
|
user=test_user,
|
|
invoice_number="INV-000001-202311",
|
|
period_start=date(2023, 11, 1),
|
|
period_end=date(2023, 11, 30),
|
|
subtotal=Decimal("10.00"),
|
|
tax=Decimal("0.00"),
|
|
total=Decimal("10.00"),
|
|
status="draft"
|
|
)
|
|
|
|
assert invoice.user_id == test_user.id
|
|
assert invoice.invoice_number == "INV-000001-202311"
|
|
assert invoice.total == Decimal("10.00")
|
|
await invoice.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invoice_line_item_creation(test_user):
|
|
invoice = await Invoice.create(
|
|
user=test_user,
|
|
invoice_number="INV-000002-202311",
|
|
period_start=date(2023, 11, 1),
|
|
period_end=date(2023, 11, 30),
|
|
subtotal=Decimal("10.00"),
|
|
tax=Decimal("0.00"),
|
|
total=Decimal("10.00"),
|
|
status="draft"
|
|
)
|
|
|
|
line_item = await InvoiceLineItem.create(
|
|
invoice=invoice,
|
|
description="Storage usage",
|
|
quantity=Decimal("100.000000"),
|
|
unit_price=Decimal("0.100000"),
|
|
amount=Decimal("10.0000"),
|
|
item_type="storage"
|
|
)
|
|
|
|
assert line_item.invoice_id == invoice.id
|
|
assert line_item.description == "Storage usage"
|
|
assert line_item.amount == Decimal("10.0000")
|
|
await line_item.delete()
|
|
await invoice.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pricing_config_creation(test_user):
|
|
config = await PricingConfig.create(
|
|
config_key="storage_per_gb_month",
|
|
config_value=Decimal("0.0045"),
|
|
description="Storage cost per GB per month",
|
|
unit="per_gb_month",
|
|
updated_by=test_user
|
|
)
|
|
|
|
assert config.config_key == "storage_per_gb_month"
|
|
assert config.config_value == Decimal("0.0045")
|
|
await config.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_payment_method_creation(test_user):
|
|
payment_method = await PaymentMethod.create(
|
|
user=test_user,
|
|
stripe_payment_method_id="pm_test_123",
|
|
type="card",
|
|
is_default=True,
|
|
last4="4242",
|
|
brand="visa",
|
|
exp_month=12,
|
|
exp_year=2025
|
|
)
|
|
|
|
assert payment_method.user_id == test_user.id
|
|
assert payment_method.last4 == "4242"
|
|
assert payment_method.is_default is True
|
|
await payment_method.delete()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_billing_event_creation(test_user):
|
|
event = await BillingEvent.create(
|
|
user=test_user,
|
|
event_type="invoice_created",
|
|
stripe_event_id="evt_test_123",
|
|
data={"invoice_id": 1},
|
|
processed=False
|
|
)
|
|
|
|
assert event.user_id == test_user.id
|
|
assert event.event_type == "invoice_created"
|
|
assert event.processed is False
|
|
await event.delete()
|