refactor: replace module-scoped Tortoise fixtures with session-level conftest initialization and pytest_asyncio fixtures
Migrate billing test suite from per-module Tortoise initializer/finalizer fixtures to a shared session-scoped setup in tests/conftest.py, converting all @pytest.fixture async helpers to @pytest_asyncio.fixture and removing the deleted e2e Playwright test directory.
This commit is contained in:
@@ -1,28 +1,15 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime
|
||||
from httpx import AsyncClient
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from fastapi import status
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from mywebdav.main import app
|
||||
from mywebdav.models import User
|
||||
from mywebdav.billing.models import PricingConfig, Invoice, UsageAggregate, UserSubscription
|
||||
from mywebdav.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(["mywebdav.models", "mywebdav.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user():
|
||||
user = await User.create(
|
||||
username="testuser",
|
||||
@@ -34,7 +21,7 @@ async def test_user():
|
||||
yield user
|
||||
await user.delete()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def admin_user():
|
||||
user = await User.create(
|
||||
username="adminuser",
|
||||
@@ -46,17 +33,17 @@ async def admin_user():
|
||||
yield user
|
||||
await user.delete()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def auth_token(test_user):
|
||||
token = create_access_token(data={"sub": test_user.username})
|
||||
return token
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def admin_token(admin_user):
|
||||
token = create_access_token(data={"sub": admin_user.username})
|
||||
return token
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def pricing_config():
|
||||
configs = []
|
||||
configs.append(await PricingConfig.create(
|
||||
@@ -89,7 +76,7 @@ async def pricing_config():
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_current_usage(test_user, auth_token):
|
||||
async with AsyncClient(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}"}
|
||||
@@ -113,7 +100,7 @@ async def test_get_monthly_usage(test_user, auth_token):
|
||||
bandwidth_down_bytes=1024 ** 3 * 5
|
||||
)
|
||||
|
||||
async with AsyncClient(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}"}
|
||||
@@ -127,7 +114,7 @@ async def test_get_monthly_usage(test_user, auth_token):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_subscription(test_user, auth_token):
|
||||
async with AsyncClient(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}"}
|
||||
@@ -153,7 +140,7 @@ async def test_list_invoices(test_user, auth_token):
|
||||
status="open"
|
||||
)
|
||||
|
||||
async with AsyncClient(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}"}
|
||||
@@ -179,7 +166,7 @@ async def test_get_invoice(test_user, auth_token):
|
||||
status="open"
|
||||
)
|
||||
|
||||
async with AsyncClient(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}"}
|
||||
@@ -193,7 +180,7 @@ async def test_get_invoice(test_user, auth_token):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pricing():
|
||||
async with AsyncClient(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
|
||||
@@ -202,7 +189,7 @@ async def test_get_pricing():
|
||||
|
||||
@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:
|
||||
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}"}
|
||||
@@ -216,7 +203,7 @@ async def test_admin_get_pricing(admin_user, admin_token, pricing_config):
|
||||
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:
|
||||
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}"},
|
||||
@@ -233,7 +220,7 @@ async def test_admin_update_pricing(admin_user, admin_token, pricing_config):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_get_stats(admin_user, admin_token):
|
||||
async with AsyncClient(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}"}
|
||||
@@ -247,7 +234,7 @@ async def test_admin_get_stats(admin_user, admin_token):
|
||||
|
||||
@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:
|
||||
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}"}
|
||||
|
||||
@@ -1,25 +1,12 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from mywebdav.models import User
|
||||
from mywebdav.billing.models import Invoice, InvoiceLineItem, PricingConfig, UsageAggregate, UserSubscription
|
||||
from mywebdav.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(["mywebdav.models", "mywebdav.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user():
|
||||
user = await User.create(
|
||||
username="testuser",
|
||||
@@ -30,7 +17,7 @@ async def test_user():
|
||||
yield user
|
||||
await user.delete()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def pricing_config():
|
||||
configs = []
|
||||
configs.append(await PricingConfig.create(
|
||||
@@ -170,8 +157,14 @@ async def test_mark_invoice_paid(test_user, pricing_config):
|
||||
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"))
|
||||
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"))
|
||||
assert updated == 1 # Should update 1 row
|
||||
|
||||
# Verify the update worked
|
||||
tax_config = await PricingConfig.get(config_key="tax_rate_default")
|
||||
assert tax_config.config_value == Decimal("0.21")
|
||||
|
||||
today = date.today()
|
||||
|
||||
|
||||
@@ -2,26 +2,12 @@ import pytest
|
||||
import pytest_asyncio
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from mywebdav.models import User
|
||||
from mywebdav.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(["mywebdav.models", "mywebdav.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user():
|
||||
user = await User.create(
|
||||
|
||||
@@ -1,25 +1,12 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from decimal import Decimal
|
||||
from datetime import date, datetime, timedelta
|
||||
from tortoise.contrib.test import initializer, finalizer
|
||||
from mywebdav.models import User, File, Folder
|
||||
from mywebdav.billing.models import UsageRecord, UsageAggregate
|
||||
from mywebdav.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(["mywebdav.models", "mywebdav.billing.models"], db_url="sqlite://:memory:")
|
||||
yield
|
||||
await finalizer()
|
||||
|
||||
@pytest.fixture
|
||||
@pytest_asyncio.fixture
|
||||
async def test_user():
|
||||
user = await User.create(
|
||||
username="testuser",
|
||||
|
||||
Reference in New Issue
Block a user