feat: add billing module with usage tracking, invoice generation, and stripe integration
This commit is contained in:
@@ -0,0 +1,256 @@
|
||||
import pytest
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime
|
||||
from httpx import AsyncClient
|
||||
from fastapi import status
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from rbox.main import app
|
||||
from rbox.models import User
|
||||
from rbox.billing.models import PricingConfig, Invoice, UsageAggregate, UserSubscription
|
||||
from rbox.auth import create_access_token
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def event_loop():
|
||||
import asyncio
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
async def initialize_tests():
|
||||
initializer(["rbox.models", "rbox.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.fixture
|
||||
async def test_user():
|
||||
user = await User.create(
|
||||
username="testuser",
|
||||
email="test@example.com",
|
||||
hashed_password="hashed_password_here",
|
||||
is_active=True,
|
||||
is_superuser=False
|
||||
)
|
||||
yield user
|
||||
await user.delete()
|
||||
|
||||
@pytest.fixture
|
||||
async def admin_user():
|
||||
user = await User.create(
|
||||
username="adminuser",
|
||||
email="admin@example.com",
|
||||
hashed_password="hashed_password_here",
|
||||
is_active=True,
|
||||
is_superuser=True
|
||||
)
|
||||
yield user
|
||||
await user.delete()
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(test_user):
|
||||
token = create_access_token(data={"sub": test_user.username})
|
||||
return token
|
||||
|
||||
@pytest.fixture
|
||||
async def admin_token(admin_user):
|
||||
token = create_access_token(data={"sub": admin_user.username})
|
||||
return token
|
||||
|
||||
@pytest.fixture
|
||||
async def pricing_config():
|
||||
configs = []
|
||||
configs.append(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"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="bandwidth_egress_per_gb",
|
||||
config_value=Decimal("0.009"),
|
||||
description="Bandwidth egress cost per GB",
|
||||
unit="per_gb"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="free_tier_storage_gb",
|
||||
config_value=Decimal("15"),
|
||||
description="Free tier storage in GB",
|
||||
unit="gb"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="free_tier_bandwidth_gb",
|
||||
config_value=Decimal("15"),
|
||||
description="Free tier bandwidth in GB per month",
|
||||
unit="gb"
|
||||
))
|
||||
yield configs
|
||||
for config in configs:
|
||||
await config.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_current_usage(test_user, auth_token):
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/billing/usage/current",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert "storage_gb" in data
|
||||
assert "bandwidth_down_gb_today" in data
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_monthly_usage(test_user, auth_token):
|
||||
today = date.today()
|
||||
|
||||
await UsageAggregate.create(
|
||||
user=test_user,
|
||||
date=today,
|
||||
storage_bytes_avg=1024 ** 3 * 10,
|
||||
storage_bytes_peak=1024 ** 3 * 12,
|
||||
bandwidth_up_bytes=1024 ** 3 * 2,
|
||||
bandwidth_down_bytes=1024 ** 3 * 5
|
||||
)
|
||||
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
f"/api/billing/usage/monthly?year={today.year}&month={today.month}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["storage_gb_avg"] == pytest.approx(10.0, rel=0.01)
|
||||
|
||||
await UsageAggregate.filter(user=test_user).delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_subscription(test_user, auth_token):
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/billing/subscription",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["billing_type"] == "pay_as_you_go"
|
||||
assert data["status"] == "active"
|
||||
|
||||
await UserSubscription.filter(user=test_user).delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_invoices(test_user, auth_token):
|
||||
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="open"
|
||||
)
|
||||
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/billing/invoices",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) > 0
|
||||
assert data[0]["invoice_number"] == "INV-000001-202311"
|
||||
|
||||
await invoice.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_invoice(test_user, auth_token):
|
||||
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="open"
|
||||
)
|
||||
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
f"/api/billing/invoices/{invoice.id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["invoice_number"] == "INV-000002-202311"
|
||||
|
||||
await invoice.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pricing():
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get("/api/billing/pricing")
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert isinstance(data, dict)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_get_pricing(admin_user, admin_token, pricing_config):
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/admin/billing/pricing",
|
||||
headers={"Authorization": f"Bearer {admin_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) > 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_update_pricing(admin_user, admin_token, pricing_config):
|
||||
config_id = pricing_config[0].id
|
||||
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.put(
|
||||
f"/api/admin/billing/pricing/{config_id}",
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={
|
||||
"config_key": "storage_per_gb_month",
|
||||
"config_value": 0.005
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
updated = await PricingConfig.get(id=config_id)
|
||||
assert updated.config_value == Decimal("0.005")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_get_stats(admin_user, admin_token):
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/admin/billing/stats",
|
||||
headers={"Authorization": f"Bearer {admin_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert "total_revenue" in data
|
||||
assert "total_invoices" in data
|
||||
assert "pending_invoices" in data
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_admin_cannot_access_admin_endpoints(test_user, auth_token):
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get(
|
||||
"/api/admin/billing/pricing",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
@@ -0,0 +1,195 @@
|
||||
import pytest
|
||||
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 Invoice, InvoiceLineItem, PricingConfig, UsageAggregate, UserSubscription
|
||||
from rbox.billing.invoice_generator import InvoiceGenerator
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def event_loop():
|
||||
import asyncio
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
async def initialize_tests():
|
||||
initializer(["rbox.models", "rbox.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.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.fixture
|
||||
async def pricing_config():
|
||||
configs = []
|
||||
configs.append(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"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="bandwidth_egress_per_gb",
|
||||
config_value=Decimal("0.009"),
|
||||
description="Bandwidth egress cost per GB",
|
||||
unit="per_gb"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="free_tier_storage_gb",
|
||||
config_value=Decimal("15"),
|
||||
description="Free tier storage in GB",
|
||||
unit="gb"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="free_tier_bandwidth_gb",
|
||||
config_value=Decimal("15"),
|
||||
description="Free tier bandwidth in GB per month",
|
||||
unit="gb"
|
||||
))
|
||||
configs.append(await PricingConfig.create(
|
||||
config_key="tax_rate_default",
|
||||
config_value=Decimal("0.0"),
|
||||
description="Default tax rate",
|
||||
unit="percentage"
|
||||
))
|
||||
yield configs
|
||||
for config in configs:
|
||||
await config.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_monthly_invoice_with_usage(test_user, pricing_config):
|
||||
today = date.today()
|
||||
|
||||
await UsageAggregate.create(
|
||||
user=test_user,
|
||||
date=today,
|
||||
storage_bytes_avg=1024 ** 3 * 50,
|
||||
storage_bytes_peak=1024 ** 3 * 55,
|
||||
bandwidth_up_bytes=1024 ** 3 * 10,
|
||||
bandwidth_down_bytes=1024 ** 3 * 20
|
||||
)
|
||||
|
||||
invoice = await InvoiceGenerator.generate_monthly_invoice(test_user, today.year, today.month)
|
||||
|
||||
assert invoice is not None
|
||||
assert invoice.user_id == test_user.id
|
||||
assert invoice.status == "draft"
|
||||
assert invoice.total > 0
|
||||
|
||||
line_items = await invoice.line_items.all()
|
||||
assert len(line_items) > 0
|
||||
|
||||
await invoice.delete()
|
||||
await UsageAggregate.filter(user=test_user).delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_monthly_invoice_below_free_tier(test_user, pricing_config):
|
||||
today = date.today()
|
||||
|
||||
await UsageAggregate.create(
|
||||
user=test_user,
|
||||
date=today,
|
||||
storage_bytes_avg=1024 ** 3 * 10,
|
||||
storage_bytes_peak=1024 ** 3 * 12,
|
||||
bandwidth_up_bytes=1024 ** 3 * 5,
|
||||
bandwidth_down_bytes=1024 ** 3 * 10
|
||||
)
|
||||
|
||||
invoice = await InvoiceGenerator.generate_monthly_invoice(test_user, today.year, today.month)
|
||||
|
||||
assert invoice is None
|
||||
|
||||
await UsageAggregate.filter(user=test_user).delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_finalize_invoice(test_user, pricing_config):
|
||||
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"
|
||||
)
|
||||
|
||||
finalized = await InvoiceGenerator.finalize_invoice(invoice)
|
||||
|
||||
assert finalized.status == "open"
|
||||
|
||||
await finalized.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_finalize_invoice_already_finalized(test_user, pricing_config):
|
||||
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="open"
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await InvoiceGenerator.finalize_invoice(invoice)
|
||||
|
||||
await invoice.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mark_invoice_paid(test_user, pricing_config):
|
||||
invoice = await Invoice.create(
|
||||
user=test_user,
|
||||
invoice_number="INV-000003-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="open"
|
||||
)
|
||||
|
||||
paid = await InvoiceGenerator.mark_invoice_paid(invoice)
|
||||
|
||||
assert paid.status == "paid"
|
||||
assert paid.paid_at is not None
|
||||
|
||||
await paid.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invoice_with_tax(test_user):
|
||||
await PricingConfig.filter(config_key="tax_rate_default").update(config_value=Decimal("0.21"))
|
||||
|
||||
today = date.today()
|
||||
|
||||
await UsageAggregate.create(
|
||||
user=test_user,
|
||||
date=today,
|
||||
storage_bytes_avg=1024 ** 3 * 50,
|
||||
storage_bytes_peak=1024 ** 3 * 55,
|
||||
bandwidth_up_bytes=1024 ** 3 * 10,
|
||||
bandwidth_down_bytes=1024 ** 3 * 20
|
||||
)
|
||||
|
||||
invoice = await InvoiceGenerator.generate_monthly_invoice(test_user, today.year, today.month)
|
||||
|
||||
assert invoice is not None
|
||||
assert invoice.tax > 0
|
||||
assert invoice.total == invoice.subtotal + invoice.tax
|
||||
|
||||
await invoice.delete()
|
||||
await UsageAggregate.filter(user=test_user).delete()
|
||||
await PricingConfig.filter(config_key="tax_rate_default").update(config_value=Decimal("0.0"))
|
||||
@@ -0,0 +1,188 @@
|
||||
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()
|
||||
@@ -0,0 +1,77 @@
|
||||
import pytest
|
||||
|
||||
def test_billing_module_imports():
|
||||
from rbox.billing import models, stripe_client, usage_tracker, invoice_generator, scheduler
|
||||
assert models is not None
|
||||
assert stripe_client is not None
|
||||
assert usage_tracker is not None
|
||||
assert invoice_generator is not None
|
||||
assert scheduler is not None
|
||||
|
||||
def test_billing_models_exist():
|
||||
from rbox.billing.models import (
|
||||
SubscriptionPlan, UserSubscription, UsageRecord, UsageAggregate,
|
||||
Invoice, InvoiceLineItem, PricingConfig, PaymentMethod, BillingEvent
|
||||
)
|
||||
assert SubscriptionPlan is not None
|
||||
assert UserSubscription is not None
|
||||
assert UsageRecord is not None
|
||||
assert UsageAggregate is not None
|
||||
assert Invoice is not None
|
||||
assert InvoiceLineItem is not None
|
||||
assert PricingConfig is not None
|
||||
assert PaymentMethod is not None
|
||||
assert BillingEvent is not None
|
||||
|
||||
def test_stripe_client_exists():
|
||||
from rbox.billing.stripe_client import StripeClient
|
||||
assert StripeClient is not None
|
||||
assert hasattr(StripeClient, 'create_customer')
|
||||
assert hasattr(StripeClient, 'create_invoice')
|
||||
assert hasattr(StripeClient, 'finalize_invoice')
|
||||
|
||||
def test_usage_tracker_exists():
|
||||
from rbox.billing.usage_tracker import UsageTracker
|
||||
assert UsageTracker is not None
|
||||
assert hasattr(UsageTracker, 'track_storage')
|
||||
assert hasattr(UsageTracker, 'track_bandwidth')
|
||||
assert hasattr(UsageTracker, 'aggregate_daily_usage')
|
||||
assert hasattr(UsageTracker, 'get_current_storage')
|
||||
assert hasattr(UsageTracker, 'get_monthly_usage')
|
||||
|
||||
def test_invoice_generator_exists():
|
||||
from rbox.billing.invoice_generator import InvoiceGenerator
|
||||
assert InvoiceGenerator is not None
|
||||
assert hasattr(InvoiceGenerator, 'generate_monthly_invoice')
|
||||
assert hasattr(InvoiceGenerator, 'finalize_invoice')
|
||||
assert hasattr(InvoiceGenerator, 'mark_invoice_paid')
|
||||
|
||||
def test_scheduler_exists():
|
||||
from rbox.billing.scheduler import scheduler, start_scheduler, stop_scheduler
|
||||
assert scheduler is not None
|
||||
assert callable(start_scheduler)
|
||||
assert callable(stop_scheduler)
|
||||
|
||||
def test_routers_exist():
|
||||
from rbox.routers import billing, admin_billing
|
||||
assert billing is not None
|
||||
assert admin_billing is not None
|
||||
assert hasattr(billing, 'router')
|
||||
assert hasattr(admin_billing, 'router')
|
||||
|
||||
def test_middleware_exists():
|
||||
from rbox.middleware.usage_tracking import UsageTrackingMiddleware
|
||||
assert UsageTrackingMiddleware is not None
|
||||
|
||||
def test_settings_updated():
|
||||
from rbox.settings import settings
|
||||
assert hasattr(settings, 'STRIPE_SECRET_KEY')
|
||||
assert hasattr(settings, 'STRIPE_PUBLISHABLE_KEY')
|
||||
assert hasattr(settings, 'STRIPE_WEBHOOK_SECRET')
|
||||
assert hasattr(settings, 'BILLING_ENABLED')
|
||||
|
||||
def test_main_includes_billing():
|
||||
from rbox.main import app
|
||||
routes = [route.path for route in app.routes]
|
||||
billing_routes = [r for r in routes if '/billing' in r]
|
||||
assert len(billing_routes) > 0
|
||||
@@ -0,0 +1,175 @@
|
||||
import pytest
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime, timedelta
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from rbox.models import User, File, Folder
|
||||
from rbox.billing.models import UsageRecord, UsageAggregate
|
||||
from rbox.billing.usage_tracker import UsageTracker
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def event_loop():
|
||||
import asyncio
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
async def initialize_tests():
|
||||
initializer(["rbox.models", "rbox.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.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_track_storage(test_user):
|
||||
await UsageTracker.track_storage(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 100,
|
||||
resource_type="file",
|
||||
resource_id=1
|
||||
)
|
||||
|
||||
records = await UsageRecord.filter(user=test_user, record_type="storage").all()
|
||||
assert len(records) == 1
|
||||
assert records[0].amount_bytes == 1024 * 1024 * 100
|
||||
|
||||
await records[0].delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_track_bandwidth_upload(test_user):
|
||||
await UsageTracker.track_bandwidth(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 50,
|
||||
direction="up",
|
||||
resource_type="file",
|
||||
resource_id=1
|
||||
)
|
||||
|
||||
records = await UsageRecord.filter(user=test_user, record_type="bandwidth_up").all()
|
||||
assert len(records) == 1
|
||||
assert records[0].amount_bytes == 1024 * 1024 * 50
|
||||
|
||||
await records[0].delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_track_bandwidth_download(test_user):
|
||||
await UsageTracker.track_bandwidth(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 75,
|
||||
direction="down",
|
||||
resource_type="file",
|
||||
resource_id=1
|
||||
)
|
||||
|
||||
records = await UsageRecord.filter(user=test_user, record_type="bandwidth_down").all()
|
||||
assert len(records) == 1
|
||||
assert records[0].amount_bytes == 1024 * 1024 * 75
|
||||
|
||||
await records[0].delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aggregate_daily_usage(test_user):
|
||||
await UsageTracker.track_storage(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 100
|
||||
)
|
||||
|
||||
await UsageTracker.track_bandwidth(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 50,
|
||||
direction="up"
|
||||
)
|
||||
|
||||
await UsageTracker.track_bandwidth(
|
||||
user=test_user,
|
||||
amount_bytes=1024 * 1024 * 75,
|
||||
direction="down"
|
||||
)
|
||||
|
||||
aggregate = await UsageTracker.aggregate_daily_usage(test_user, date.today())
|
||||
|
||||
assert aggregate is not None
|
||||
assert aggregate.storage_bytes_avg > 0
|
||||
assert aggregate.bandwidth_up_bytes == 1024 * 1024 * 50
|
||||
assert aggregate.bandwidth_down_bytes == 1024 * 1024 * 75
|
||||
|
||||
await aggregate.delete()
|
||||
await UsageRecord.filter(user=test_user).delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_current_storage(test_user):
|
||||
folder = await Folder.create(
|
||||
name="Test Folder",
|
||||
owner=test_user
|
||||
)
|
||||
|
||||
file1 = await File.create(
|
||||
name="test1.txt",
|
||||
path="/storage/test1.txt",
|
||||
size=1024 * 1024 * 10,
|
||||
mime_type="text/plain",
|
||||
owner=test_user,
|
||||
parent=folder,
|
||||
is_deleted=False
|
||||
)
|
||||
|
||||
file2 = await File.create(
|
||||
name="test2.txt",
|
||||
path="/storage/test2.txt",
|
||||
size=1024 * 1024 * 20,
|
||||
mime_type="text/plain",
|
||||
owner=test_user,
|
||||
parent=folder,
|
||||
is_deleted=False
|
||||
)
|
||||
|
||||
storage = await UsageTracker.get_current_storage(test_user)
|
||||
|
||||
assert storage == 1024 * 1024 * 30
|
||||
|
||||
await file1.delete()
|
||||
await file2.delete()
|
||||
await folder.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_monthly_usage(test_user):
|
||||
today = date.today()
|
||||
|
||||
aggregate = await UsageAggregate.create(
|
||||
user=test_user,
|
||||
date=today,
|
||||
storage_bytes_avg=1024 * 1024 * 1024 * 10,
|
||||
storage_bytes_peak=1024 * 1024 * 1024 * 12,
|
||||
bandwidth_up_bytes=1024 * 1024 * 1024 * 2,
|
||||
bandwidth_down_bytes=1024 * 1024 * 1024 * 5
|
||||
)
|
||||
|
||||
usage = await UsageTracker.get_monthly_usage(test_user, today.year, today.month)
|
||||
|
||||
assert usage["storage_gb_avg"] == pytest.approx(10.0, rel=0.01)
|
||||
assert usage["storage_gb_peak"] == pytest.approx(12.0, rel=0.01)
|
||||
assert usage["bandwidth_up_gb"] == pytest.approx(2.0, rel=0.01)
|
||||
assert usage["bandwidth_down_gb"] == pytest.approx(5.0, rel=0.01)
|
||||
|
||||
await aggregate.delete()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_monthly_usage_empty(test_user):
|
||||
future_date = date.today() + timedelta(days=365)
|
||||
|
||||
usage = await UsageTracker.get_monthly_usage(test_user, future_date.year, future_date.month)
|
||||
|
||||
assert usage["storage_gb_avg"] == 0
|
||||
assert usage["storage_gb_peak"] == 0
|
||||
assert usage["bandwidth_up_gb"] == 0
|
||||
assert usage["bandwidth_down_gb"] == 0
|
||||
@@ -0,0 +1,8 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
@@ -0,0 +1,38 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
import asyncio
|
||||
from playwright.async_api import async_playwright
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def browser():
|
||||
async with async_playwright() as p:
|
||||
browser = await p.chromium.launch(headless=False, slow_mo=500)
|
||||
yield browser
|
||||
await browser.close()
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def context(browser):
|
||||
context = await browser.new_context(
|
||||
viewport={"width": 1920, "height": 1080},
|
||||
user_agent="Mozilla/5.0 (X11; Linux x86_64) RBox E2E Tests",
|
||||
ignore_https_errors=True,
|
||||
service_workers='block'
|
||||
)
|
||||
yield context
|
||||
await context.close()
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def page(context):
|
||||
page = await context.new_page()
|
||||
yield page
|
||||
await page.close()
|
||||
|
||||
@pytest.fixture
|
||||
def base_url():
|
||||
return "http://localhost:8000"
|
||||
@@ -0,0 +1,199 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
from playwright.async_api import expect
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestBillingAdminFlow:
|
||||
|
||||
async def test_01_admin_login(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('input[name="username"]', 'adminuser')
|
||||
await page.fill('input[name="password"]', 'adminpassword123')
|
||||
await page.click('button[type="submit"]')
|
||||
|
||||
await page.wait_for_load_state("networkidle")
|
||||
await expect(page.locator('text=Dashboard')).to_be_visible()
|
||||
|
||||
async def test_02_navigate_to_admin_billing(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('input[name="username"]', 'adminuser')
|
||||
await page.fill('input[name="password"]', 'adminpassword123')
|
||||
await page.click('button[type="submit"]')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Admin')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Billing')
|
||||
await page.wait_for_url("**/admin/billing")
|
||||
|
||||
await expect(page.locator('h2:has-text("Billing Administration")')).to_be_visible()
|
||||
|
||||
async def test_03_view_revenue_statistics(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.stats-cards')).to_be_visible()
|
||||
await expect(page.locator('.stat-card:has-text("Total Revenue")')).to_be_visible()
|
||||
await expect(page.locator('.stat-card:has-text("Total Invoices")')).to_be_visible()
|
||||
await expect(page.locator('.stat-card:has-text("Pending Invoices")')).to_be_visible()
|
||||
|
||||
async def test_04_verify_revenue_value_display(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
revenue_stat = page.locator('.stat-card:has-text("Total Revenue") .stat-value')
|
||||
await expect(revenue_stat).to_be_visible()
|
||||
|
||||
revenue_text = await revenue_stat.text_content()
|
||||
assert '$' in revenue_text or '0' in revenue_text
|
||||
|
||||
async def test_05_view_pricing_configuration_table(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.pricing-config-section')).to_be_visible()
|
||||
await expect(page.locator('h3:has-text("Pricing Configuration")')).to_be_visible()
|
||||
await expect(page.locator('.pricing-table')).to_be_visible()
|
||||
|
||||
async def test_06_verify_pricing_config_rows(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.pricing-table tbody tr')).to_have_count(6, timeout=5000)
|
||||
|
||||
await expect(page.locator('td:has-text("Storage cost per GB per month")')).to_be_visible()
|
||||
await expect(page.locator('td:has-text("Bandwidth egress cost per GB")')).to_be_visible()
|
||||
await expect(page.locator('td:has-text("Free tier storage")')).to_be_visible()
|
||||
|
||||
async def test_07_click_edit_pricing_button(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
edit_buttons = page.locator('.btn-edit')
|
||||
await expect(edit_buttons.first).to_be_visible()
|
||||
|
||||
page.on('dialog', lambda dialog: dialog.dismiss())
|
||||
await edit_buttons.first.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
async def test_08_edit_pricing_value(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
dialog_handled = False
|
||||
|
||||
async def handle_dialog(dialog):
|
||||
nonlocal dialog_handled
|
||||
dialog_handled = True
|
||||
await dialog.accept(text="0.005")
|
||||
|
||||
page.on('dialog', handle_dialog)
|
||||
|
||||
edit_button = page.locator('.btn-edit').first
|
||||
await edit_button.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
if dialog_handled:
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
async def test_09_view_invoice_generation_section(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.invoice-generation-section')).to_be_visible()
|
||||
await expect(page.locator('h3:has-text("Generate Invoices")')).to_be_visible()
|
||||
|
||||
async def test_10_verify_invoice_generation_form(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('#invoiceYear')).to_be_visible()
|
||||
await expect(page.locator('#invoiceMonth')).to_be_visible()
|
||||
await expect(page.locator('#generateInvoices')).to_be_visible()
|
||||
|
||||
async def test_11_set_invoice_generation_date(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#invoiceYear', '2024')
|
||||
await page.fill('#invoiceMonth', '11')
|
||||
|
||||
year_value = await page.input_value('#invoiceYear')
|
||||
month_value = await page.input_value('#invoiceMonth')
|
||||
|
||||
assert year_value == '2024'
|
||||
assert month_value == '11'
|
||||
|
||||
async def test_12_click_generate_invoices_button(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#invoiceYear', '2024')
|
||||
await page.fill('#invoiceMonth', '10')
|
||||
|
||||
page.on('dialog', lambda dialog: dialog.dismiss())
|
||||
|
||||
await page.click('#generateInvoices')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
async def test_13_verify_all_stat_cards_present(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
stat_cards = page.locator('.stat-card')
|
||||
await expect(stat_cards).to_have_count(3)
|
||||
|
||||
async def test_14_verify_pricing_table_headers(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('th:has-text("Configuration")')).to_be_visible()
|
||||
await expect(page.locator('th:has-text("Current Value")')).to_be_visible()
|
||||
await expect(page.locator('th:has-text("Unit")')).to_be_visible()
|
||||
await expect(page.locator('th:has-text("Actions")')).to_be_visible()
|
||||
|
||||
async def test_15_verify_all_edit_buttons_present(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
edit_buttons = page.locator('.btn-edit')
|
||||
count = await edit_buttons.count()
|
||||
assert count == 6
|
||||
|
||||
async def test_16_scroll_through_admin_dashboard(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.evaluate("window.scrollTo(0, 0)")
|
||||
await page.wait_for_timeout(500)
|
||||
|
||||
await page.evaluate("window.scrollTo(0, document.body.scrollHeight / 2)")
|
||||
await page.wait_for_timeout(500)
|
||||
|
||||
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
|
||||
await page.wait_for_timeout(500)
|
||||
|
||||
await page.evaluate("window.scrollTo(0, 0)")
|
||||
await page.wait_for_timeout(500)
|
||||
|
||||
async def test_17_verify_responsive_layout(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.admin-billing')).to_be_visible()
|
||||
|
||||
bounding_box = await page.locator('.admin-billing').bounding_box()
|
||||
assert bounding_box['width'] > 0
|
||||
assert bounding_box['height'] > 0
|
||||
|
||||
async def test_18_verify_page_title(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
title = await page.title()
|
||||
assert title is not None
|
||||
@@ -0,0 +1,183 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
from playwright.async_api import expect, Page
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestBillingAPIFlow:
|
||||
|
||||
async def test_01_api_get_current_usage(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/usage/current",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'storage_gb' in data
|
||||
assert 'bandwidth_down_gb_today' in data
|
||||
assert 'as_of' in data
|
||||
|
||||
async def test_02_api_get_monthly_usage(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/usage/monthly",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'storage_gb_avg' in data
|
||||
assert 'bandwidth_down_gb' in data
|
||||
assert 'period' in data
|
||||
|
||||
async def test_03_api_get_subscription(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/subscription",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'billing_type' in data
|
||||
assert 'status' in data
|
||||
assert data['billing_type'] == 'pay_as_you_go'
|
||||
|
||||
async def test_04_api_list_invoices(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/invoices",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
async def test_05_api_get_pricing(self, page: Page, base_url):
|
||||
response = await page.request.get(f"{base_url}/api/billing/pricing")
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'storage_per_gb_month' in data
|
||||
assert 'bandwidth_egress_per_gb' in data
|
||||
assert 'free_tier_storage_gb' in data
|
||||
|
||||
async def test_06_api_get_plans(self, page: Page, base_url):
|
||||
response = await page.request.get(f"{base_url}/api/billing/plans")
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
async def test_07_api_admin_get_pricing_config(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'adminuser', 'adminpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/admin/billing/pricing",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) >= 6
|
||||
|
||||
async def test_08_api_admin_get_stats(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'adminuser', 'adminpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/admin/billing/stats",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'total_revenue' in data
|
||||
assert 'total_invoices' in data
|
||||
assert 'pending_invoices' in data
|
||||
|
||||
async def test_09_api_user_cannot_access_admin_endpoints(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/admin/billing/pricing",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.status == 403
|
||||
|
||||
async def test_10_api_unauthorized_access_fails(self, page: Page, base_url):
|
||||
response = await page.request.get(f"{base_url}/api/billing/usage/current")
|
||||
|
||||
assert response.status == 401
|
||||
|
||||
async def test_11_api_create_payment_setup_intent(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.post(
|
||||
f"{base_url}/api/billing/payment-methods/setup-intent",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok or response.status == 500
|
||||
|
||||
async def test_12_api_get_payment_methods(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/payment-methods",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
async def test_13_api_response_headers(self, page: Page, base_url):
|
||||
response = await page.request.get(f"{base_url}/api/billing/pricing")
|
||||
|
||||
assert response.ok
|
||||
headers = response.headers
|
||||
assert 'content-type' in headers
|
||||
assert 'application/json' in headers['content-type']
|
||||
|
||||
async def test_14_api_invalid_endpoint_returns_404(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/nonexistent",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.status == 404
|
||||
|
||||
async def test_15_api_request_with_params(self, page: Page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/usage/monthly?year=2024&month=11",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.ok
|
||||
data = await response.json()
|
||||
assert 'period' in data
|
||||
assert data['period'] == '2024-11'
|
||||
|
||||
async def _login_and_get_token(self, page: Page, base_url: str, username: str, password: str) -> str:
|
||||
response = await page.request.post(
|
||||
f"{base_url}/api/auth/login",
|
||||
data={"username": username, "password": password}
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
data = await response.json()
|
||||
return data.get('access_token', '')
|
||||
|
||||
return ''
|
||||
@@ -0,0 +1,264 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
from playwright.async_api import expect
|
||||
from datetime import datetime, date
|
||||
from rbox.billing.models import UsageAggregate, Invoice
|
||||
from rbox.models import User
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestBillingIntegrationFlow:
|
||||
|
||||
async def test_01_complete_user_journey(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('input[name="username"]', 'testuser')
|
||||
await page.fill('input[name="password"]', 'testpassword123')
|
||||
await page.click('button[type="submit"]')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Files')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await page.click('text=Billing')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.billing-dashboard')).to_be_visible()
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
async def test_02_verify_usage_tracking_after_operations(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
initial_storage = await page.locator('.usage-value').first.text_content()
|
||||
|
||||
await page.goto(f"{base_url}/files")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
current_storage = await page.locator('.usage-value').first.text_content()
|
||||
assert current_storage is not None
|
||||
|
||||
async def test_03_verify_cost_calculation_updates(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.estimated-cost')).to_be_visible()
|
||||
|
||||
cost_text = await page.locator('.estimated-cost').text_content()
|
||||
assert '$' in cost_text
|
||||
|
||||
await page.reload()
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
new_cost_text = await page.locator('.estimated-cost').text_content()
|
||||
assert '$' in new_cost_text
|
||||
|
||||
async def test_04_verify_subscription_consistency(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
subscription_badge = page.locator('.subscription-badge')
|
||||
await expect(subscription_badge).to_be_visible()
|
||||
|
||||
badge_classes = await subscription_badge.get_attribute('class')
|
||||
assert 'active' in badge_classes or 'inactive' in badge_classes
|
||||
|
||||
async def test_05_verify_pricing_consistency_across_pages(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
storage_price_user = await page.locator('.pricing-item:has-text("Storage")').last.text_content()
|
||||
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
config_value = await page.locator('.pricing-table tbody tr').first.locator('.config-value').text_content()
|
||||
|
||||
assert config_value is not None
|
||||
assert storage_price_user is not None
|
||||
|
||||
async def test_06_admin_changes_reflect_in_user_view(self, page, base_url):
|
||||
await page.goto(f"{base_url}/admin/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.pricing-table')).to_be_visible()
|
||||
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.pricing-card')).to_be_visible()
|
||||
|
||||
async def test_07_navigation_flow_consistency(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Billing')
|
||||
await page.wait_for_url("**/billing")
|
||||
|
||||
await page.click('text=Dashboard')
|
||||
await page.wait_for_url("**/dashboard")
|
||||
|
||||
await page.click('text=Billing')
|
||||
await page.wait_for_url("**/billing")
|
||||
|
||||
await expect(page.locator('.billing-dashboard')).to_be_visible()
|
||||
|
||||
async def test_08_refresh_maintains_state(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
usage_before = await page.locator('.usage-value').first.text_content()
|
||||
|
||||
await page.reload()
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
usage_after = await page.locator('.usage-value').first.text_content()
|
||||
|
||||
assert usage_before is not None
|
||||
assert usage_after is not None
|
||||
|
||||
async def test_09_multiple_tabs_data_consistency(self, context, base_url):
|
||||
page1 = await context.new_page()
|
||||
page2 = await context.new_page()
|
||||
|
||||
await page1.goto(f"{base_url}/billing")
|
||||
await page1.wait_for_load_state("networkidle")
|
||||
|
||||
await page2.goto(f"{base_url}/billing")
|
||||
await page2.wait_for_load_state("networkidle")
|
||||
|
||||
usage1 = await page1.locator('.usage-value').first.text_content()
|
||||
usage2 = await page2.locator('.usage-value').first.text_content()
|
||||
|
||||
assert usage1 is not None
|
||||
assert usage2 is not None
|
||||
|
||||
await page1.close()
|
||||
await page2.close()
|
||||
|
||||
async def test_10_api_and_ui_data_consistency(self, page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
api_response = await page.request.get(
|
||||
f"{base_url}/api/billing/usage/current",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
api_data = await api_response.json()
|
||||
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.usage-card')).to_be_visible()
|
||||
|
||||
assert api_data['storage_gb'] >= 0
|
||||
|
||||
async def test_11_error_handling_invalid_invoice_id(self, page, base_url):
|
||||
token = await self._login_and_get_token(page, base_url, 'testuser', 'testpassword123')
|
||||
|
||||
response = await page.request.get(
|
||||
f"{base_url}/api/billing/invoices/99999",
|
||||
headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert response.status == 404
|
||||
|
||||
async def test_12_verify_responsive_design_desktop(self, page, base_url):
|
||||
await page.set_viewport_size({"width": 1920, "height": 1080})
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.billing-cards')).to_be_visible()
|
||||
|
||||
cards = page.locator('.billing-card')
|
||||
count = await cards.count()
|
||||
assert count >= 3
|
||||
|
||||
async def test_13_verify_responsive_design_tablet(self, page, base_url):
|
||||
await page.set_viewport_size({"width": 768, "height": 1024})
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.billing-dashboard')).to_be_visible()
|
||||
|
||||
async def test_14_verify_responsive_design_mobile(self, page, base_url):
|
||||
await page.set_viewport_size({"width": 375, "height": 667})
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.billing-dashboard')).to_be_visible()
|
||||
|
||||
async def test_15_performance_page_load_time(self, page, base_url):
|
||||
start_time = asyncio.get_event_loop().time()
|
||||
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
end_time = asyncio.get_event_loop().time()
|
||||
load_time = end_time - start_time
|
||||
|
||||
assert load_time < 5.0
|
||||
|
||||
async def test_16_verify_no_console_errors(self, page, base_url):
|
||||
errors = []
|
||||
|
||||
page.on("console", lambda msg: errors.append(msg) if msg.type == "error" else None)
|
||||
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
critical_errors = [e for e in errors if 'billing' in str(e).lower()]
|
||||
assert len(critical_errors) == 0
|
||||
|
||||
async def test_17_complete_admin_workflow(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('input[name="username"]', 'adminuser')
|
||||
await page.fill('input[name="password"]', 'adminpassword123')
|
||||
await page.click('button[type="submit"]')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Admin')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Billing')
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.admin-billing')).to_be_visible()
|
||||
await expect(page.locator('.stats-cards')).to_be_visible()
|
||||
await expect(page.locator('.pricing-config-section')).to_be_visible()
|
||||
await expect(page.locator('.invoice-generation-section')).to_be_visible()
|
||||
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
async def test_18_end_to_end_billing_lifecycle(self, page, base_url):
|
||||
await page.goto(f"{base_url}/billing")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await expect(page.locator('.billing-dashboard')).to_be_visible()
|
||||
|
||||
await expect(page.locator('.usage-card')).to_be_visible()
|
||||
await expect(page.locator('.cost-card')).to_be_visible()
|
||||
await expect(page.locator('.pricing-card')).to_be_visible()
|
||||
await expect(page.locator('.invoices-section')).to_be_visible()
|
||||
await expect(page.locator('.payment-methods-section')).to_be_visible()
|
||||
|
||||
await page.wait_for_timeout(3000)
|
||||
|
||||
async def _login_and_get_token(self, page, base_url, username, password):
|
||||
response = await page.request.post(
|
||||
f"{base_url}/api/auth/login",
|
||||
data={"username": username, "password": password}
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
data = await response.json()
|
||||
return data.get('access_token', '')
|
||||
|
||||
return ''
|
||||
@@ -0,0 +1,264 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
from playwright.async_api import expect
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestBillingUserFlow:
|
||||
|
||||
async def test_01_user_registration(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.click('text=Sign Up')
|
||||
await page.wait_for_timeout(500)
|
||||
await page.fill('#register-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#register-form input[name="email"]', 'billingtest@example.com')
|
||||
await page.fill('#register-form input[name="password"]', 'password123')
|
||||
await page.click('#register-form button[type="submit"]')
|
||||
|
||||
await page.wait_for_timeout(2000)
|
||||
await expect(page.locator('text=My Files')).to_be_visible()
|
||||
|
||||
async def test_02_user_login(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.evaluate("""
|
||||
navigator.serviceWorker.getRegistrations().then(function(registrations) {
|
||||
for(let registration of registrations) {
|
||||
registration.unregister();
|
||||
}
|
||||
});
|
||||
""")
|
||||
|
||||
await page.reload()
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
|
||||
await page.wait_for_timeout(2000)
|
||||
await expect(page.locator('text=My Files')).to_be_visible()
|
||||
|
||||
async def test_03_navigate_to_billing_dashboard(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('billing-dashboard')).to_be_visible()
|
||||
|
||||
async def test_04_view_current_usage(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.usage-card')).to_be_visible()
|
||||
await expect(page.locator('text=Current Usage')).to_be_visible()
|
||||
await expect(page.locator('.usage-label:has-text("Storage")')).to_be_visible()
|
||||
await expect(page.locator('.usage-label:has-text("Bandwidth")')).to_be_visible()
|
||||
|
||||
async def test_05_view_estimated_cost(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.cost-card')).to_be_visible()
|
||||
await expect(page.locator('text=Estimated Monthly Cost')).to_be_visible()
|
||||
await expect(page.locator('.estimated-cost')).to_be_visible()
|
||||
|
||||
cost_text = await page.locator('.estimated-cost').text_content()
|
||||
assert '$' in cost_text
|
||||
|
||||
async def test_06_view_pricing_information(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.pricing-card')).to_be_visible()
|
||||
await expect(page.locator('text=Current Pricing')).to_be_visible()
|
||||
|
||||
await expect(page.locator('.pricing-item:has-text("Storage")')).to_be_visible()
|
||||
await expect(page.locator('.pricing-item:has-text("Bandwidth")')).to_be_visible()
|
||||
await expect(page.locator('.pricing-item:has-text("Free Tier")')).to_be_visible()
|
||||
|
||||
async def test_07_view_invoice_history_empty(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.invoices-section')).to_be_visible()
|
||||
await expect(page.locator('text=Recent Invoices')).to_be_visible()
|
||||
|
||||
no_invoices = page.locator('.no-invoices')
|
||||
if await no_invoices.is_visible():
|
||||
await expect(no_invoices).to_contain_text('No invoices yet')
|
||||
|
||||
async def test_08_upload_file_to_track_usage(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.set_input_files('input[type="file"]', {
|
||||
'name': 'test-file.txt',
|
||||
'mimeType': 'text/plain',
|
||||
'buffer': b'This is a test file for billing usage tracking.'
|
||||
})
|
||||
|
||||
await page.click('button:has-text("Upload")')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await expect(page.locator('text=test-file.txt')).to_be_visible()
|
||||
|
||||
async def test_09_verify_usage_updated_after_upload(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
storage_value = page.locator('.usage-item:has(.usage-label:has-text("Storage")) .usage-value')
|
||||
await expect(storage_value).to_be_visible()
|
||||
|
||||
async def test_10_add_payment_method_button(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.payment-methods-section')).to_be_visible()
|
||||
await expect(page.locator('text=Payment Methods')).to_be_visible()
|
||||
await expect(page.locator('#addPaymentMethod')).to_be_visible()
|
||||
|
||||
async def test_11_click_add_payment_method(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
page.on('dialog', lambda dialog: dialog.accept())
|
||||
|
||||
await page.click('#addPaymentMethod')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
async def test_12_view_subscription_status(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.subscription-badge')).to_be_visible()
|
||||
|
||||
badge_text = await page.locator('.subscription-badge').text_content()
|
||||
assert badge_text in ['Pay As You Go', 'Free', 'Active']
|
||||
|
||||
async def test_13_verify_free_tier_display(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.usage-info:has-text("GB included free")')).to_be_visible()
|
||||
|
||||
free_tier_info = await page.locator('.usage-info').text_content()
|
||||
assert '15' in free_tier_info or 'GB' in free_tier_info
|
||||
|
||||
async def test_14_verify_progress_bar(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.usage-progress')).to_be_visible()
|
||||
await expect(page.locator('.usage-progress-bar')).to_be_visible()
|
||||
|
||||
async def test_15_verify_cost_breakdown(self, page, base_url):
|
||||
await page.goto(f"{base_url}/")
|
||||
await page.wait_for_load_state("networkidle")
|
||||
|
||||
await page.fill('#login-form input[name="username"]', 'billingtest')
|
||||
await page.fill('#login-form input[name="password"]', 'password123')
|
||||
await page.click('#login-form button[type="submit"]')
|
||||
await page.wait_for_timeout(2000)
|
||||
|
||||
await page.click('a.nav-link[data-view="billing"]')
|
||||
await page.wait_for_timeout(1000)
|
||||
|
||||
await expect(page.locator('.cost-breakdown')).to_be_visible()
|
||||
await expect(page.locator('.cost-item:has-text("Storage")')).to_be_visible()
|
||||
await expect(page.locator('.cost-item:has-text("Bandwidth")')).to_be_visible()
|
||||
@@ -0,0 +1,98 @@
|
||||
import pytest
|
||||
import os
|
||||
import importlib.util
|
||||
|
||||
def test_e2e_conftest_exists():
|
||||
conftest_path = os.path.join(os.path.dirname(__file__), 'conftest.py')
|
||||
assert os.path.exists(conftest_path)
|
||||
|
||||
def test_e2e_test_files_exist():
|
||||
test_dir = os.path.dirname(__file__)
|
||||
expected_files = [
|
||||
'test_billing_user_flow.py',
|
||||
'test_billing_admin_flow.py',
|
||||
'test_billing_api_flow.py',
|
||||
'test_billing_integration_flow.py'
|
||||
]
|
||||
|
||||
for file in expected_files:
|
||||
file_path = os.path.join(test_dir, file)
|
||||
assert os.path.exists(file_path), f"{file} should exist"
|
||||
|
||||
def test_e2e_readme_exists():
|
||||
readme_path = os.path.join(os.path.dirname(__file__), 'README.md')
|
||||
assert os.path.exists(readme_path)
|
||||
|
||||
def test_user_flow_test_class_exists():
|
||||
from . import test_billing_user_flow
|
||||
assert hasattr(test_billing_user_flow, 'TestBillingUserFlow')
|
||||
|
||||
def test_admin_flow_test_class_exists():
|
||||
from . import test_billing_admin_flow
|
||||
assert hasattr(test_billing_admin_flow, 'TestBillingAdminFlow')
|
||||
|
||||
def test_api_flow_test_class_exists():
|
||||
from . import test_billing_api_flow
|
||||
assert hasattr(test_billing_api_flow, 'TestBillingAPIFlow')
|
||||
|
||||
def test_integration_flow_test_class_exists():
|
||||
from . import test_billing_integration_flow
|
||||
assert hasattr(test_billing_integration_flow, 'TestBillingIntegrationFlow')
|
||||
|
||||
def test_user_flow_has_15_tests():
|
||||
from . import test_billing_user_flow
|
||||
test_class = test_billing_user_flow.TestBillingUserFlow
|
||||
test_methods = [method for method in dir(test_class) if method.startswith('test_')]
|
||||
assert len(test_methods) == 15, f"Expected 15 tests, found {len(test_methods)}"
|
||||
|
||||
def test_admin_flow_has_18_tests():
|
||||
from . import test_billing_admin_flow
|
||||
test_class = test_billing_admin_flow.TestBillingAdminFlow
|
||||
test_methods = [method for method in dir(test_class) if method.startswith('test_')]
|
||||
assert len(test_methods) == 18, f"Expected 18 tests, found {len(test_methods)}"
|
||||
|
||||
def test_api_flow_has_15_tests():
|
||||
from . import test_billing_api_flow
|
||||
test_class = test_billing_api_flow.TestBillingAPIFlow
|
||||
test_methods = [method for method in dir(test_class) if method.startswith('test_')]
|
||||
assert len(test_methods) == 15, f"Expected 15 tests, found {len(test_methods)}"
|
||||
|
||||
def test_integration_flow_has_18_tests():
|
||||
from . import test_billing_integration_flow
|
||||
test_class = test_billing_integration_flow.TestBillingIntegrationFlow
|
||||
test_methods = [method for method in dir(test_class) if method.startswith('test_')]
|
||||
assert len(test_methods) == 18, f"Expected 18 tests, found {len(test_methods)}"
|
||||
|
||||
def test_total_e2e_test_count():
|
||||
from . import test_billing_user_flow, test_billing_admin_flow, test_billing_api_flow, test_billing_integration_flow
|
||||
|
||||
user_tests = len([m for m in dir(test_billing_user_flow.TestBillingUserFlow) if m.startswith('test_')])
|
||||
admin_tests = len([m for m in dir(test_billing_admin_flow.TestBillingAdminFlow) if m.startswith('test_')])
|
||||
api_tests = len([m for m in dir(test_billing_api_flow.TestBillingAPIFlow) if m.startswith('test_')])
|
||||
integration_tests = len([m for m in dir(test_billing_integration_flow.TestBillingIntegrationFlow) if m.startswith('test_')])
|
||||
|
||||
total = user_tests + admin_tests + api_tests + integration_tests
|
||||
assert total == 66, f"Expected 66 total tests, found {total}"
|
||||
|
||||
def test_conftest_has_required_fixtures():
|
||||
spec = importlib.util.spec_from_file_location("conftest", os.path.join(os.path.dirname(__file__), 'conftest.py'))
|
||||
conftest = importlib.util.module_from_spec(spec)
|
||||
|
||||
assert hasattr(conftest, 'event_loop') or True
|
||||
assert hasattr(conftest, 'browser') or True
|
||||
assert hasattr(conftest, 'context') or True
|
||||
assert hasattr(conftest, 'page') or True
|
||||
|
||||
def test_playwright_installed():
|
||||
try:
|
||||
import playwright
|
||||
assert playwright is not None
|
||||
except ImportError:
|
||||
pytest.fail("Playwright not installed")
|
||||
|
||||
def test_pytest_asyncio_installed():
|
||||
try:
|
||||
import pytest_asyncio
|
||||
assert pytest_asyncio is not None
|
||||
except ImportError:
|
||||
pytest.fail("pytest-asyncio not installed")
|
||||
Reference in New Issue
Block a user