139 lines
5.4 KiB
Python
139 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from asgiref.sync import sync_to_async
|
|
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseServerError
|
|
from django.test import AsyncClient, Client
|
|
from shop.models import Product
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_search_product():
|
|
# 新增商品
|
|
product = await sync_to_async(Product.objects.create)(
|
|
name="测试商品", description="描述", price="9.99", stock=10, keywords="测试,商品"
|
|
)
|
|
# 搜索商品
|
|
client = AsyncClient()
|
|
response = await client.get("/shop/product/search/", {"q": "测试"})
|
|
assert response.status_code == HttpResponse.status_code
|
|
data = response.json()
|
|
assert "products" in data
|
|
found = any(str(product.id) == str(p.get("id")) or product.name == p.get("name") for p in data["products"])
|
|
assert found
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_update_product_view():
|
|
# 新增商品
|
|
product = Product.objects.create(name="待更新商品", description="desc", price="19.99", stock=5, keywords="更新")
|
|
client = Client()
|
|
url = "/shop/product/update/"
|
|
update_data = {"id": product.id, "data": {"name": "已更新商品", "price": "29.99"}}
|
|
response = client.post(url, data=json.dumps(update_data), content_type="application/json")
|
|
assert response.status_code == HttpResponse.status_code
|
|
product.refresh_from_db()
|
|
assert product.name == "已更新商品"
|
|
assert str(product.price) == "29.99"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_update_product_view_invalid():
|
|
client = Client()
|
|
url = "/shop/product/update/"
|
|
# 缺少id
|
|
response = client.post(url, data=json.dumps({"data": {"name": "x"}}), content_type="application/json")
|
|
assert response.status_code == HttpResponseBadRequest.status_code
|
|
# 缺少data
|
|
response = client.post(url, data=json.dumps({"id": 1}), content_type="application/json")
|
|
assert response.status_code == HttpResponseBadRequest.status_code
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_delete_product():
|
|
# 新增商品
|
|
product = Product.objects.create(name="待删除商品", description="desc", price="5.00", stock=1, keywords="删除")
|
|
pid = product.id
|
|
product.delete()
|
|
assert not Product.objects.filter(id=pid).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_update_product_empty_fields():
|
|
client = Client()
|
|
response = client.post("/shop/product/update/", data="{}", content_type="application/json")
|
|
assert response.status_code == HttpResponseBadRequest.status_code
|
|
assert "商品ID和更新数据不能为空" in response.json().get("error", "")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.asyncio
|
|
async def test_update_product_success():
|
|
client = AsyncClient()
|
|
with patch(
|
|
"shop.services.ProductService.update_product", new_callable=AsyncMock, return_value={"updated": True}
|
|
) as mock_update:
|
|
data = {"id": 1, "data": {"name": "新商品名"}}
|
|
response = await client.post("/shop/product/update/", data=json.dumps(data), content_type="application/json")
|
|
assert response.status_code == HttpResponse.status_code
|
|
assert response.json()["result"] == {"updated": True}
|
|
mock_update.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.asyncio
|
|
async def test_update_product_exception():
|
|
client = AsyncClient()
|
|
with patch(
|
|
"shop.services.ProductService.update_product", new_callable=AsyncMock, side_effect=Exception("mock error")
|
|
):
|
|
data = {"id": 1, "data": {"name": "新商品名"}}
|
|
response = await client.post("/shop/product/update/", data=json.dumps(data), content_type="application/json")
|
|
assert response.status_code == HttpResponseServerError.status_code
|
|
assert response.json()["error"] == "系统异常"
|
|
assert "mock error" in response.json().get("detail", "")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_batch_order_empty_items():
|
|
client = Client()
|
|
response = client.post("/shop/order/batch/", data="{}", content_type="application/json")
|
|
assert response.status_code == HttpResponseBadRequest.status_code
|
|
assert "订单明细不能为空" in response.json().get("error", "")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_batch_order_success():
|
|
client = Client()
|
|
order_id = 123
|
|
with patch(
|
|
"shop.services.OrderService.batch_create_order",
|
|
return_value=(
|
|
order_id,
|
|
[
|
|
{"product_id": 1, "status": "success"},
|
|
],
|
|
),
|
|
) as mock_batch:
|
|
data = {"items": [{"product_id": 1, "quantity": 2}]}
|
|
response = client.post("/shop/order/batch/", data=json.dumps(data), content_type="application/json")
|
|
assert response.status_code == HttpResponse.status_code
|
|
assert response.json()["order_id"] == order_id
|
|
assert response.json()["results"] == [{"product_id": 1, "status": "success"}]
|
|
mock_batch.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_batch_order_exception():
|
|
client = Client()
|
|
with patch("shop.services.OrderService.batch_create_order", side_effect=Exception("mock error")):
|
|
data = {"items": [{"product_id": 1, "quantity": 2}]}
|
|
response = client.post("/shop/order/batch/", data=json.dumps(data), content_type="application/json")
|
|
assert response.status_code == HttpResponseServerError.status_code
|
|
assert response.json()["error"] == "系统异常"
|
|
assert "mock error" in response.json().get("detail", "")
|