Spaces:
Paused
Paused
File size: 4,883 Bytes
4c5298a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
pytestmark = pytest.mark.asyncio
@pytest.fixture
async def user_token(client: AsyncClient) -> str:
"""Create a test user and return their token"""
# Register user
await client.post(
"/api/v1/auth/register",
data={
"username": "testuser@example.com",
"password": "testpass123",
"full_name": "Test User"
}
)
# Login to get token
response = await client.post(
"/api/v1/auth/login",
data={
"username": "testuser@example.com",
"password": "testpass123"
}
)
return response.json()["access_token"]
@pytest.fixture
async def test_product(client: AsyncClient, user_token: str) -> dict:
"""Create a test product and return its data"""
response = await client.post(
"/api/v1/products/",
json={
"name": "Test Product",
"description": "A test product",
"price": 99.99,
"category": "test",
"inventory_count": 10
},
headers={"Authorization": f"Bearer {user_token}"}
)
return response.json()
async def test_create_order(client: AsyncClient, user_token: str, test_product: dict):
# Test creating an order
response = await client.post(
"/api/v1/orders/",
json={
"items": [
{
"product_id": test_product["id"],
"quantity": 2
}
]
},
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 1
assert data["items"][0]["product_id"] == test_product["id"]
assert data["total_amount"] == test_product["price"] * 2
return data["id"]
async def test_get_order(client: AsyncClient, user_token: str, test_product: dict):
# First create an order
order_id = await test_create_order(client, user_token, test_product)
# Test getting the order
response = await client.get(
f"/api/v1/orders/{order_id}",
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["id"] == order_id
assert len(data["items"]) == 1
async def test_list_orders(client: AsyncClient, user_token: str, test_product: dict):
# Create multiple orders
await test_create_order(client, user_token, test_product)
await test_create_order(client, user_token, test_product)
# Test listing orders
response = await client.get(
"/api/v1/orders/",
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 2
async def test_update_order_status(client: AsyncClient, user_token: str, test_product: dict):
# First create an order
order_id = await test_create_order(client, user_token, test_product)
# Test updating the order status
response = await client.put(
f"/api/v1/orders/{order_id}/status",
json={"status": "processing"},
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "processing"
# Test invalid status
response = await client.put(
f"/api/v1/orders/{order_id}/status",
json={"status": "invalid_status"},
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 400
async def test_cancel_order(client: AsyncClient, user_token: str, test_product: dict):
# First create an order
order_id = await test_create_order(client, user_token, test_product)
# Get initial product inventory
product_response = await client.get(
f"/api/v1/products/{test_product['id']}",
headers={"Authorization": f"Bearer {user_token}"}
)
initial_inventory = product_response.json()["inventory_count"]
# Cancel the order
response = await client.put(
f"/api/v1/orders/{order_id}/status",
json={"status": "cancelled"},
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == 200
assert response.json()["status"] == "cancelled"
# Verify inventory was restored
product_response = await client.get(
f"/api/v1/products/{test_product['id']}",
headers={"Authorization": f"Bearer {user_token}"}
)
final_inventory = product_response.json()["inventory_count"]
assert final_inventory == initial_inventory + 2 # We ordered 2 items |