rbox/tests/billing/test_models.py

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()