retoor/rbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: reformat code style with consistent quotes, trailing commas, and spacing across auth, billing, and models

Browse Source
This commit is contained in:
retoor
2025-11-13 22:22:05 +00:00
parent aac0798305
commit 69f3161eec
34 changed files with 1851 additions and 875 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/billing/test_api_endpoints.py
+100 -56
View File
@@ -6,9 +6,15 @@ from httpx import AsyncClient, ASGITransport
from fastapi import status
from mywebdav.main import app
from mywebdav.models import User
from mywebdav.billing.models import PricingConfig, Invoice, UsageAggregate, UserSubscription
from mywebdav.billing.models import (
PricingConfig,
Invoice,
UsageAggregate,
UserSubscription,
)
from mywebdav.auth import create_access_token
@pytest_asyncio.fixture
async def test_user():
user = await User.create(
@@ -16,11 +22,12 @@ async def test_user():
email="test@example.com",
hashed_password="hashed_password_here",
is_active=True,
is_superuser=False
is_superuser=False,
)
yield user
await user.delete()
@pytest_asyncio.fixture
async def admin_user():
user = await User.create(
@@ -28,58 +35,72 @@ async def admin_user():
email="admin@example.com",
hashed_password="hashed_password_here",
is_active=True,
is_superuser=True
is_superuser=True,
)
yield user
await user.delete()
@pytest_asyncio.fixture
async def auth_token(test_user):
token = create_access_token(data={"sub": test_user.username})
return token
@pytest_asyncio.fixture
async def admin_token(admin_user):
token = create_access_token(data={"sub": admin_user.username})
return token
@pytest_asyncio.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="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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/billing/usage/current",
headers={"Authorization": f"Bearer {auth_token}"}
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == status.HTTP_200_OK
@@ -87,6 +108,7 @@ async def test_get_current_usage(test_user, auth_token):
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()
@@ -94,16 +116,18 @@ async def test_get_monthly_usage(test_user, auth_token):
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
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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(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}"}
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == status.HTTP_200_OK
@@ -112,12 +136,15 @@ async def test_get_monthly_usage(test_user, auth_token):
await UsageAggregate.filter(user=test_user).delete()
@pytest.mark.asyncio
async def test_get_subscription(test_user, auth_token):
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/billing/subscription",
headers={"Authorization": f"Bearer {auth_token}"}
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == status.HTTP_200_OK
@@ -127,6 +154,7 @@ async def test_get_subscription(test_user, auth_token):
await UserSubscription.filter(user=test_user).delete()
@pytest.mark.asyncio
async def test_list_invoices(test_user, auth_token):
invoice = await Invoice.create(
@@ -137,13 +165,14 @@ async def test_list_invoices(test_user, auth_token):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="open"
status="open",
)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/billing/invoices",
headers={"Authorization": f"Bearer {auth_token}"}
"/api/billing/invoices", headers={"Authorization": f"Bearer {auth_token}"}
)
assert response.status_code == status.HTTP_200_OK
@@ -153,6 +182,7 @@ async def test_list_invoices(test_user, auth_token):
await invoice.delete()
@pytest.mark.asyncio
async def test_get_invoice(test_user, auth_token):
invoice = await Invoice.create(
@@ -163,13 +193,15 @@ async def test_get_invoice(test_user, auth_token):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="open"
status="open",
)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
f"/api/billing/invoices/{invoice.id}",
headers={"Authorization": f"Bearer {auth_token}"}
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == status.HTTP_200_OK
@@ -178,39 +210,45 @@ async def test_get_invoice(test_user, auth_token):
await invoice.delete()
@pytest.mark.asyncio
async def test_get_pricing():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/admin/billing/pricing",
headers={"Authorization": f"Bearer {admin_token}"}
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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(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
}
json={"config_key": "storage_per_gb_month", "config_value": 0.005},
)
assert response.status_code == status.HTTP_200_OK
@@ -218,12 +256,15 @@ async def test_admin_update_pricing(admin_user, admin_token, pricing_config):
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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/admin/billing/stats",
headers={"Authorization": f"Bearer {admin_token}"}
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == status.HTTP_200_OK
@@ -232,12 +273,15 @@ async def test_admin_get_stats(admin_user, admin_token):
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(transport=ASGITransport(app=app), base_url="http://test") as client:
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get(
"/api/admin/billing/pricing",
headers={"Authorization": f"Bearer {auth_token}"}
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == status.HTTP_403_FORBIDDEN
tests/billing/test_invoice_generator.py
+86 -52
View File
@@ -3,57 +3,76 @@ import pytest_asyncio
from decimal import Decimal
from datetime import date, datetime
from mywebdav.models import User
from mywebdav.billing.models import Invoice, InvoiceLineItem, PricingConfig, UsageAggregate, UserSubscription
from mywebdav.billing.models import (
Invoice,
InvoiceLineItem,
PricingConfig,
UsageAggregate,
UserSubscription,
)
from mywebdav.billing.invoice_generator import InvoiceGenerator
@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
is_active=True,
)
yield user
await user.delete()
@pytest_asyncio.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"
))
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()
@@ -61,13 +80,15 @@ async def test_generate_monthly_invoice_with_usage(test_user, pricing_config):
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
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)
invoice = await InvoiceGenerator.generate_monthly_invoice(
test_user, today.year, today.month
)
assert invoice is not None
assert invoice.user_id == test_user.id
@@ -80,6 +101,7 @@ async def test_generate_monthly_invoice_with_usage(test_user, pricing_config):
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()
@@ -87,18 +109,21 @@ async def test_generate_monthly_invoice_below_free_tier(test_user, pricing_confi
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
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)
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(
@@ -109,7 +134,7 @@ async def test_finalize_invoice(test_user, pricing_config):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="draft"
status="draft",
)
finalized = await InvoiceGenerator.finalize_invoice(invoice)
@@ -118,6 +143,7 @@ async def test_finalize_invoice(test_user, pricing_config):
await finalized.delete()
@pytest.mark.asyncio
async def test_finalize_invoice_already_finalized(test_user, pricing_config):
invoice = await Invoice.create(
@@ -128,7 +154,7 @@ async def test_finalize_invoice_already_finalized(test_user, pricing_config):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="open"
status="open",
)
with pytest.raises(ValueError):
@@ -136,6 +162,7 @@ async def test_finalize_invoice_already_finalized(test_user, pricing_config):
await invoice.delete()
@pytest.mark.asyncio
async def test_mark_invoice_paid(test_user, pricing_config):
invoice = await Invoice.create(
@@ -146,7 +173,7 @@ async def test_mark_invoice_paid(test_user, pricing_config):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="open"
status="open",
)
paid = await InvoiceGenerator.mark_invoice_paid(invoice)
@@ -156,10 +183,13 @@ async def test_mark_invoice_paid(test_user, pricing_config):
await paid.delete()
@pytest.mark.asyncio
async def test_invoice_with_tax(test_user, pricing_config):
# Update tax rate
updated = await PricingConfig.filter(config_key="tax_rate_default").update(config_value=Decimal("0.21"))
updated = await PricingConfig.filter(config_key="tax_rate_default").update(
config_value=Decimal("0.21")
)
assert updated == 1 # Should update 1 row
# Verify the update worked
@@ -171,13 +201,15 @@ async def test_invoice_with_tax(test_user, pricing_config):
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
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)
invoice = await InvoiceGenerator.generate_monthly_invoice(
test_user, today.year, today.month
)
assert invoice is not None
assert invoice.tax > 0
@@ -185,4 +217,6 @@ async def test_invoice_with_tax(test_user, pricing_config):
await invoice.delete()
await UsageAggregate.filter(user=test_user).delete()
await PricingConfig.filter(config_key="tax_rate_default").update(config_value=Decimal("0.0"))
await PricingConfig.filter(config_key="tax_rate_default").update(
config_value=Decimal("0.0")
)
tests/billing/test_models.py
+30 -15
View File
@@ -4,21 +4,30 @@ from decimal import Decimal
from datetime import date, datetime
from mywebdav.models import User
from mywebdav.billing.models import (
SubscriptionPlan, UserSubscription, UsageRecord, UsageAggregate,
Invoice, InvoiceLineItem, PricingConfig, PaymentMethod, BillingEvent
SubscriptionPlan,
UserSubscription,
UsageRecord,
UsageAggregate,
Invoice,
InvoiceLineItem,
PricingConfig,
PaymentMethod,
BillingEvent,
)
@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
is_active=True,
)
yield user
await user.delete()
@pytest.mark.asyncio
async def test_subscription_plan_creation():
plan = await SubscriptionPlan.create(
@@ -28,7 +37,7 @@ async def test_subscription_plan_creation():
storage_gb=100,
bandwidth_gb=100,
price_monthly=Decimal("5.00"),
price_yearly=Decimal("50.00")
price_yearly=Decimal("50.00"),
)
assert plan.name == "starter"
@@ -36,12 +45,11 @@ async def test_subscription_plan_creation():
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"
user=test_user, billing_type="pay_as_you_go", status="active"
)
assert subscription.user_id == test_user.id
@@ -49,6 +57,7 @@ async def test_user_subscription_creation(test_user):
assert subscription.status == "active"
await subscription.delete()
@pytest.mark.asyncio
async def test_usage_record_creation(test_user):
usage = await UsageRecord.create(
@@ -57,7 +66,7 @@ async def test_usage_record_creation(test_user):
amount_bytes=1024 * 1024 * 100,
resource_type="file",
resource_id=1,
idempotency_key="test_key_123"
idempotency_key="test_key_123",
)
assert usage.user_id == test_user.id
@@ -65,6 +74,7 @@ async def test_usage_record_creation(test_user):
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(
@@ -73,13 +83,14 @@ async def test_usage_aggregate_creation(test_user):
storage_bytes_avg=1024 * 1024 * 500,
storage_bytes_peak=1024 * 1024 * 600,
bandwidth_up_bytes=1024 * 1024 * 50,
bandwidth_down_bytes=1024 * 1024 * 100
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(
@@ -90,7 +101,7 @@ async def test_invoice_creation(test_user):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="draft"
status="draft",
)
assert invoice.user_id == test_user.id
@@ -98,6 +109,7 @@ async def test_invoice_creation(test_user):
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(
@@ -108,7 +120,7 @@ async def test_invoice_line_item_creation(test_user):
subtotal=Decimal("10.00"),
tax=Decimal("0.00"),
total=Decimal("10.00"),
status="draft"
status="draft",
)
line_item = await InvoiceLineItem.create(
@@ -117,7 +129,7 @@ async def test_invoice_line_item_creation(test_user):
quantity=Decimal("100.000000"),
unit_price=Decimal("0.100000"),
amount=Decimal("10.0000"),
item_type="storage"
item_type="storage",
)
assert line_item.invoice_id == invoice.id
@@ -126,6 +138,7 @@ async def test_invoice_line_item_creation(test_user):
await line_item.delete()
await invoice.delete()
@pytest.mark.asyncio
async def test_pricing_config_creation(test_user):
config = await PricingConfig.create(
@@ -133,13 +146,14 @@ async def test_pricing_config_creation(test_user):
config_value=Decimal("0.0045"),
description="Storage cost per GB per month",
unit="per_gb_month",
updated_by=test_user
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(
@@ -150,7 +164,7 @@ async def test_payment_method_creation(test_user):
last4="4242",
brand="visa",
exp_month=12,
exp_year=2025
exp_year=2025,
)
assert payment_method.user_id == test_user.id
@@ -158,6 +172,7 @@ async def test_payment_method_creation(test_user):
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(
@@ -165,7 +180,7 @@ async def test_billing_event_creation(test_user):
event_type="invoice_created",
stripe_event_id="evt_test_123",
data={"invoice_id": 1},
processed=False
processed=False,
)
assert event.user_id == test_user.id
tests/billing/test_simple.py
+54 -21
View File
@@ -1,18 +1,35 @@
import pytest
def test_billing_module_imports():
from mywebdav.billing import models, stripe_client, usage_tracker, invoice_generator, scheduler
from mywebdav.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 mywebdav.billing.models import (
SubscriptionPlan, UserSubscription, UsageRecord, UsageAggregate,
Invoice, InvoiceLineItem, PricingConfig, PaymentMethod, BillingEvent
SubscriptionPlan,
UserSubscription,
UsageRecord,
UsageAggregate,
Invoice,
InvoiceLineItem,
PricingConfig,
PaymentMethod,
BillingEvent,
)
assert SubscriptionPlan is not None
assert UserSubscription is not None
assert UsageRecord is not None
@@ -23,55 +40,71 @@ def test_billing_models_exist():
assert PaymentMethod is not None
assert BillingEvent is not None
def test_stripe_client_exists():
from mywebdav.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')
assert hasattr(StripeClient, "create_customer")
assert hasattr(StripeClient, "create_invoice")
assert hasattr(StripeClient, "finalize_invoice")
def test_usage_tracker_exists():
from mywebdav.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')
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 mywebdav.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')
assert hasattr(InvoiceGenerator, "generate_monthly_invoice")
assert hasattr(InvoiceGenerator, "finalize_invoice")
assert hasattr(InvoiceGenerator, "mark_invoice_paid")
def test_scheduler_exists():
from mywebdav.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 mywebdav.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')
assert hasattr(billing, "router")
assert hasattr(admin_billing, "router")
def test_middleware_exists():
from mywebdav.middleware.usage_tracking import UsageTrackingMiddleware
assert UsageTrackingMiddleware is not None
def test_settings_updated():
from mywebdav.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')
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 mywebdav.main import app
routes = [route.path for route in app.routes]
billing_routes = [r for r in routes if '/billing' in r]
billing_routes = [r for r in routes if "/billing" in r]
assert len(billing_routes) > 0
tests/billing/test_usage_tracker.py
+27 -25
View File
@@ -6,24 +6,26 @@ from mywebdav.models import User, File, Folder
from mywebdav.billing.models import UsageRecord, UsageAggregate
from mywebdav.billing.usage_tracker import UsageTracker
@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
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
resource_id=1,
)
records = await UsageRecord.filter(user=test_user, record_type="storage").all()
@@ -32,6 +34,7 @@ async def test_track_storage(test_user):
await records[0].delete()
@pytest.mark.asyncio
async def test_track_bandwidth_upload(test_user):
await UsageTracker.track_bandwidth(
@@ -39,7 +42,7 @@ async def test_track_bandwidth_upload(test_user):
amount_bytes=1024 * 1024 * 50,
direction="up",
resource_type="file",
resource_id=1
resource_id=1,
)
records = await UsageRecord.filter(user=test_user, record_type="bandwidth_up").all()
@@ -48,6 +51,7 @@ async def test_track_bandwidth_upload(test_user):
await records[0].delete()
@pytest.mark.asyncio
async def test_track_bandwidth_download(test_user):
await UsageTracker.track_bandwidth(
@@ -55,32 +59,28 @@ async def test_track_bandwidth_download(test_user):
amount_bytes=1024 * 1024 * 75,
direction="down",
resource_type="file",
resource_id=1
resource_id=1,
)
records = await UsageRecord.filter(user=test_user, record_type="bandwidth_down").all()
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_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 * 50,
direction="up"
)
await UsageTracker.track_bandwidth(
user=test_user,
amount_bytes=1024 * 1024 * 75,
direction="down"
user=test_user, amount_bytes=1024 * 1024 * 75, direction="down"
)
aggregate = await UsageTracker.aggregate_daily_usage(test_user, date.today())
@@ -93,12 +93,10 @@ async def test_aggregate_daily_usage(test_user):
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
)
folder = await Folder.create(name="Test Folder", owner=test_user)
file1 = await File.create(
name="test1.txt",
@@ -107,7 +105,7 @@ async def test_get_current_storage(test_user):
mime_type="text/plain",
owner=test_user,
parent=folder,
is_deleted=False
is_deleted=False,
)
file2 = await File.create(
@@ -117,7 +115,7 @@ async def test_get_current_storage(test_user):
mime_type="text/plain",
owner=test_user,
parent=folder,
is_deleted=False
is_deleted=False,
)
storage = await UsageTracker.get_current_storage(test_user)
@@ -128,6 +126,7 @@ async def test_get_current_storage(test_user):
await file2.delete()
await folder.delete()
@pytest.mark.asyncio
async def test_get_monthly_usage(test_user):
today = date.today()
@@ -138,7 +137,7 @@ async def test_get_monthly_usage(test_user):
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
bandwidth_down_bytes=1024 * 1024 * 1024 * 5,
)
usage = await UsageTracker.get_monthly_usage(test_user, today.year, today.month)
@@ -150,11 +149,14 @@ async def test_get_monthly_usage(test_user):
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)
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
tests/conftest.py
+12 -18
View File
@@ -3,27 +3,21 @@ import pytest_asyncio
import asyncio
from tortoise import Tortoise
# Initialize Tortoise at module level
async def _init_tortoise():
@pytest_asyncio.fixture(scope="session")
async def init_db():
"""Initialize the database for testing."""
await Tortoise.init(
db_url="sqlite://:memory:",
modules={
"models": ["mywebdav.models"],
"billing": ["mywebdav.billing.models"]
}
modules={"models": ["mywebdav.models"], "billing": ["mywebdav.billing.models"]},
)
await Tortoise.generate_schemas()
# Run initialization
asyncio.run(_init_tortoise())
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="session", autouse=True)
async def cleanup_db():
yield
await Tortoise.close_connections()
@pytest_asyncio.fixture(autouse=True)
async def setup_db(init_db):
"""Ensure database is initialized before each test."""
# The init_db fixture handles the setup
yield