219 lines
8.0 KiB
Python
219 lines
8.0 KiB
Python
"""inventory 集成测试。"""
|
||
|
||
import pytest
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
from model_bakery import baker
|
||
from rest_framework.test import APIClient
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
from apps.inventory.models import Warehouse, Stock, StockBatch, StockMovement
|
||
from apps.inventory import services as inv_services
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_client(db, user, tenant):
|
||
c = APIClient()
|
||
refresh = RefreshToken.for_user(user)
|
||
c.credentials(
|
||
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
|
||
HTTP_X_TENANT_ID=tenant.code,
|
||
)
|
||
return c
|
||
|
||
|
||
@pytest.fixture
|
||
def warehouse(db, tenant):
|
||
return baker.make(Warehouse, tenant=tenant, code="WH01", name="主仓")
|
||
|
||
|
||
@pytest.fixture
|
||
def product(db, tenant):
|
||
from apps.catalog.models import Product
|
||
return baker.make(Product, tenant=tenant, code="P001", name="测试商品")
|
||
|
||
|
||
# --- services 单元测试（不依赖 API） ---
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_inbound_creates_stock_record(tenant, warehouse, product):
|
||
result = inv_services.inbound(
|
||
tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"), unit_cost=Decimal("5"),
|
||
)
|
||
stock = Stock.objects.get(tenant=tenant, warehouse=warehouse, product=product)
|
||
assert stock.on_hand == Decimal("100")
|
||
assert stock.avg_cost == Decimal("5")
|
||
assert result.movement.movement_type == "inbound"
|
||
assert result.movement.quantity == Decimal("100")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_inbound_weighted_avg_cost(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("10"), unit_cost=Decimal("10"))
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("10"), unit_cost=Decimal("20"))
|
||
stock = Stock.objects.get(tenant=tenant, warehouse=warehouse, product=product)
|
||
# (10*10 + 20*10) / 20 = 15
|
||
assert stock.avg_cost == Decimal("15.0000")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_outbound_decreases_on_hand(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("30"))
|
||
stock = Stock.objects.get(tenant=tenant, warehouse=warehouse, product=product)
|
||
assert stock.on_hand == Decimal("70")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_outbound_insufficient_raises(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("10"))
|
||
with pytest.raises(inv_services.InsufficientStock):
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("20"))
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_outbound_no_stock_record_raises(tenant, warehouse, product):
|
||
with pytest.raises(inv_services.InsufficientStock):
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("1"))
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_lock_and_unlock(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
inv_services.lock(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("40"))
|
||
stock = Stock.objects.get(tenant=tenant, warehouse=warehouse, product=product)
|
||
assert stock.on_hand == Decimal("100")
|
||
assert stock.locked == Decimal("40")
|
||
|
||
# 锁定后再出 70 应失败（可用 = 100 - 40 = 60 < 70）
|
||
with pytest.raises(inv_services.InsufficientStock):
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("70"))
|
||
# 但出 60 可以
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("60"))
|
||
|
||
inv_services.unlock(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("20"))
|
||
stock.refresh_from_db()
|
||
assert stock.on_hand == Decimal("40")
|
||
assert stock.locked == Decimal("20")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_adjust_sets_on_hand(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
inv_services.adjust(tenant=tenant, warehouse=warehouse, product=product,
|
||
new_on_hand=Decimal("85"))
|
||
stock = Stock.objects.get(tenant=tenant, warehouse=warehouse, product=product)
|
||
assert stock.on_hand == Decimal("85")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_movements_recorded(tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("30"))
|
||
movements = StockMovement.objects.filter(tenant=tenant).order_by("id")
|
||
assert movements.count() == 2
|
||
assert movements[0].movement_type == "inbound"
|
||
assert movements[1].movement_type == "outbound"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_negative_quantity_rejected(tenant, warehouse, product):
|
||
with pytest.raises(ValueError):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("-1"))
|
||
with pytest.raises(ValueError):
|
||
inv_services.outbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("0"))
|
||
|
||
|
||
# --- API 集成测试 ---
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_api_create_warehouse(auth_client, tenant):
|
||
resp = auth_client.post(
|
||
"/api/v1/inventory/warehouses/",
|
||
{"code": "WH02", "name": "二号仓", "is_default": False},
|
||
format="json",
|
||
)
|
||
assert resp.status_code == 201, resp.content
|
||
assert resp.json()["code"] == "WH02"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_api_list_warehouses(auth_client, tenant):
|
||
baker.make(Warehouse, tenant=tenant, code="WH1")
|
||
baker.make(Warehouse, tenant=tenant, code="WH2")
|
||
resp = auth_client.get("/api/v1/inventory/warehouses/")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["count"] == 2
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_api_list_stocks_readonly(auth_client, tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
resp = auth_client.get("/api/v1/inventory/stocks/")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["count"] == 1
|
||
assert data["results"][0]["on_hand"] == "100.0000"
|
||
assert data["results"][0]["available"] == "100.0000"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_api_list_movements(auth_client, tenant, warehouse, product):
|
||
inv_services.inbound(tenant=tenant, warehouse=warehouse, product=product,
|
||
quantity=Decimal("100"))
|
||
resp = auth_client.get("/api/v1/inventory/movements/")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["count"] == 1
|
||
assert resp.json()["results"][0]["movement_type"] == "inbound"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_api_near_expiry_filter_is_server_side(auth_client, tenant, warehouse, product):
|
||
today = date.today()
|
||
baker.make(
|
||
StockBatch,
|
||
tenant=tenant,
|
||
warehouse=warehouse,
|
||
product=product,
|
||
batch_no="NEAR",
|
||
expiry_date=today + timedelta(days=7),
|
||
on_hand=Decimal("3"),
|
||
)
|
||
baker.make(
|
||
StockBatch,
|
||
tenant=tenant,
|
||
warehouse=warehouse,
|
||
product=product,
|
||
batch_no="FAR",
|
||
expiry_date=today + timedelta(days=90),
|
||
on_hand=Decimal("3"),
|
||
)
|
||
|
||
response = auth_client.get(
|
||
"/api/v1/inventory/batches/?in_stock=1&near_expiry=1&expiry_days=30"
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
assert [item["batch_no"] for item in response.json()["results"]] == ["NEAR"]
|