"""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"]