`;
@@ -38,71 +49,90 @@ export class StarredItems extends HTMLElement {
}
this.innerHTML = `
-
-
Starred Items
+
+
+
+ ${hasSelected ? `
+
+
+
+
+ ` : ''}
+
- ${this.starredFolders.map(folder => this.renderFolder(folder)).join('')}
- ${this.starredFiles.map(file => this.renderFile(file)).join('')}
+ ${this.folders.map(folder => this.renderFolder(folder)).join('')}
+ ${this.files.map(file => this.renderFile(file)).join('')}
`;
this.attachListeners();
+ this.updateIndeterminateState();
}
- renderFolder(folder) {
+ getFolderActions(folder) {
+ return `
`;
+ }
+
+ getFileActions(file) {
return `
-
+
+
`;
}
- renderFile(file) {
- const icon = this.getFileIcon(file.mime_type);
- const size = this.formatFileSize(file.size);
-
- return `
-
-
${icon}
-
${file.name}
-
${size}
-
-
+ createBatchActionsBar() {
+ const container = this.querySelector('.file-list-container');
+ const header = container.querySelector('.file-list-header');
+ const batchBar = document.createElement('div');
+ batchBar.className = 'batch-actions';
+ batchBar.innerHTML = `
+
+
`;
- }
-
- getFileIcon(mimeType) {
- if (mimeType.startsWith('image/')) return '📷';
- if (mimeType.startsWith('video/')) return '🎥';
- if (mimeType.startsWith('audio/')) return '🎵';
- if (mimeType.includes('pdf')) return '📄';
- if (mimeType.includes('text')) return '📄';
- return '📄';
- }
-
- formatFileSize(bytes) {
- if (bytes < 1024) return bytes + ' B';
- if (bytes < 1048576) return (bytes / 1024).toFixed(1) + ' KB';
- if (bytes < 1073741824) return (bytes / 1048576).toFixed(1) + ' MB';
- return (bytes / 1073741824).toFixed(1) + ' GB';
+ header.insertAdjacentElement('afterend', batchBar);
}
attachListeners() {
- this.querySelectorAll('.action-btn').forEach(btn => {
- btn.addEventListener('click', async (e) => {
- e.stopPropagation();
- const action = btn.dataset.action;
- const id = parseInt(btn.dataset.id);
- await this.handleAction(action, id);
- });
- });
+ const batchUnstarBtn = this.querySelector('#batch-unstar-btn');
+ if (batchUnstarBtn) {
+ batchUnstarBtn.addEventListener('click', () => this.handleBatchUnstar());
+ }
+ }
+
+ async handleBatchUnstar() {
+ const totalSelected = this.selectedFiles.size + this.selectedFolders.size;
+ if (totalSelected === 0) return;
+
+ if (!confirm(`Unstar ${totalSelected} items?`)) return;
+
+ try {
+ for (const fileId of this.selectedFiles) {
+ await api.unstarFile(fileId);
+ }
+ for (const folderId of this.selectedFolders) {
+ await api.unstarFolder(folderId);
+ }
+ document.dispatchEvent(new CustomEvent('show-toast', {
+ detail: { message: 'Items unstarred successfully!', type: 'success' }
+ }));
+ await this.loadStarredItems();
+ } catch (error) {
+ document.dispatchEvent(new CustomEvent('show-toast', {
+ detail: { message: 'Failed to unstar items: ' + error.message, type: 'error' }
+ }));
+ }
}
async handleAction(action, id) {
@@ -110,7 +140,7 @@ export class StarredItems extends HTMLElement {
switch (action) {
case 'download':
const blob = await api.downloadFile(id);
- const file = this.starredFiles.find(f => f.id === id);
+ const file = this.files.find(f => f.id === id);
const url = URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
diff --git a/static/js/main.js b/static/js/main.js
index b73a3a5..703e106 100644
--- a/static/js/main.js
+++ b/static/js/main.js
@@ -12,12 +12,6 @@ const app = new RBoxApp();
// Register service worker for PWA
if ('serviceWorker' in navigator) {
window.addEventListener('load', () => {
- navigator.serviceWorker.register('/static/service-worker.js')
- .then(registration => {
- console.log('Service Worker registered with scope:', registration.scope);
- })
- .catch(error => {
- console.log('Service Worker registration failed:', error);
- });
+ navigator.serviceWorker.register('/static/service-worker.js');
});
}
diff --git a/static/service-worker.js b/static/service-worker.js
index 6e11032..40f542b 100644
--- a/static/service-worker.js
+++ b/static/service-worker.js
@@ -1,4 +1,4 @@
-const CACHE_NAME = 'rbox-cache-v1';
+const CACHE_NAME = 'rbox-cache-v11';
const urlsToCache = [
'/',
'/static/index.html',
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/billing/__init__.py b/tests/billing/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/billing/test_api_endpoints.py b/tests/billing/test_api_endpoints.py
new file mode 100644
index 0000000..a10ed16
--- /dev/null
+++ b/tests/billing/test_api_endpoints.py
@@ -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
diff --git a/tests/billing/test_invoice_generator.py b/tests/billing/test_invoice_generator.py
new file mode 100644
index 0000000..d844c41
--- /dev/null
+++ b/tests/billing/test_invoice_generator.py
@@ -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"))
diff --git a/tests/billing/test_models.py b/tests/billing/test_models.py
new file mode 100644
index 0000000..6481c03
--- /dev/null
+++ b/tests/billing/test_models.py
@@ -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()
diff --git a/tests/billing/test_simple.py b/tests/billing/test_simple.py
new file mode 100644
index 0000000..2dda104
--- /dev/null
+++ b/tests/billing/test_simple.py
@@ -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
diff --git a/tests/billing/test_usage_tracker.py b/tests/billing/test_usage_tracker.py
new file mode 100644
index 0000000..d316a73
--- /dev/null
+++ b/tests/billing/test_usage_tracker.py
@@ -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
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..3d1f671
--- /dev/null
+++ b/tests/conftest.py
@@ -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()
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 0000000..1b32b11
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -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"
diff --git a/tests/e2e/test_billing_admin_flow.py b/tests/e2e/test_billing_admin_flow.py
new file mode 100644
index 0000000..3cfa59f
--- /dev/null
+++ b/tests/e2e/test_billing_admin_flow.py
@@ -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
diff --git a/tests/e2e/test_billing_api_flow.py b/tests/e2e/test_billing_api_flow.py
new file mode 100644
index 0000000..4e828b0
--- /dev/null
+++ b/tests/e2e/test_billing_api_flow.py
@@ -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 ''
diff --git a/tests/e2e/test_billing_integration_flow.py b/tests/e2e/test_billing_integration_flow.py
new file mode 100644
index 0000000..53d7e20
--- /dev/null
+++ b/tests/e2e/test_billing_integration_flow.py
@@ -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 ''
diff --git a/tests/e2e/test_billing_user_flow.py b/tests/e2e/test_billing_user_flow.py
new file mode 100644
index 0000000..37070a2
--- /dev/null
+++ b/tests/e2e/test_billing_user_flow.py
@@ -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()
diff --git a/tests/e2e/test_structure_validation.py b/tests/e2e/test_structure_validation.py
new file mode 100644
index 0000000..3d3fc36
--- /dev/null
+++ b/tests/e2e/test_structure_validation.py
@@ -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")