from datetime import timedelta from decimal import Decimal from django.contrib.auth import get_user_model from django.db.models import Sum from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from shop_management.models.data.inventory import InventoryBatch, StockBalance, StockMovement from shop_management.models.data.organization import Warehouse from shop_management.models.data.product import Product from shop_management.models.data.purchase import Purchase from shop_management.models.data.shop import Shop from shop_management.models.data.supplier import Supplier from shop_management.models.data.utils import Currency, Unit from shop_management.services.inventory import issue_stock_batch, receive_stock class PurchasePostingTests(TestCase): def setUp(self): self.user = get_user_model().objects.create_user( username="purchaser", password="test", is_staff=True ) self.user.allowed_tabs = ["purchases"] self.user.save(update_fields=["allowed_tabs"]) self.shop = Shop.objects.create(owner=self.user, name="Kabul") self.warehouse = Warehouse.objects.create( shop=self.shop, code="MAIN", name="Main" ) self.currency = Currency.objects.create(shop=self.shop, title="AFN") self.unit = Unit.objects.create(shop=self.shop, title="Box") self.supplier = Supplier.objects.create(shop=self.shop, name="Supplier") self.product = Product.objects.create( shop=self.shop, name="Medicine", default_unit=self.unit, requires_batch=True, requires_expiry=True ) self.client = APIClient() self.client.force_authenticate(self.user) def payload(self, **item_overrides): item = { "product": self.product.id, "unit": self.unit.id, "quantity": 10, "buy_price": "12.50", "retail_price": "15.00", "batch_number": "B-100", "manufacture_date": str(timezone.localdate()), "expire_date": str(timezone.localdate() + timedelta(days=365)), } item.update(item_overrides) return { "shop": self.shop.id, "supplier": self.supplier.id, "warehouse": self.warehouse.id, "currency": self.currency.id, "purchase_date": str(timezone.localdate()), "received_amount": "0", "items": [item], } def create_draft(self, payload=None): response = self.client.post( "/api/shop/purchases/", payload or self.payload(), format="json" ) self.assertEqual(response.status_code, 201, response.data) return Purchase.objects.get(pk=response.data["id"]) def test_create_is_draft_and_post_receives_inventory_once(self): purchase = self.create_draft(self.payload(buy_price="30.4040", retail_price="40.1234")) item = purchase.purchase_items.get() self.assertEqual(item.buy_price, Decimal("30.4040")) self.assertEqual(item.retail_price, Decimal("40.1234")) self.assertEqual(purchase.document_status, Purchase.DocumentStatus.DRAFT) self.assertEqual(StockMovement.objects.count(), 0) self.assertEqual(StockBalance.objects.count(), 0) response = self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") self.assertEqual(response.status_code, 200, response.data) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.POSTED) self.assertRegex(purchase.purchase_number, r"^KAB-PUR-\d{4}-\d{6}$") self.assertEqual( StockBalance.objects.get(warehouse=self.warehouse).quantity, Decimal("10.0000"), ) self.assertEqual( StockMovement.objects.filter(movement_type="PURCHASE").count(), 1 ) duplicate = self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") self.assertEqual(duplicate.status_code, 400) self.assertEqual( StockMovement.objects.filter(movement_type="PURCHASE").count(), 1 ) def test_purchase_uses_product_default_unit_automatically(self): payload = self.payload() payload["items"][0]["unit"] = None purchase = self.create_draft(payload) self.assertEqual(purchase.purchase_items.get().unit_id, self.product.default_unit_id) def test_purchase_edit_preserves_four_decimal_unit_prices(self): purchase = self.create_draft() response = self.client.put( f"/api/shop/purchases/{purchase.pk}/", self.payload(buy_price="30.4040", retail_price="40.1234"), format="json", ) self.assertEqual(response.status_code, 200, response.data) item = purchase.purchase_items.get() self.assertEqual(item.buy_price, Decimal("30.4040")) self.assertEqual(item.retail_price, Decimal("40.1234")) def test_invalid_item_does_not_partially_post(self): purchase = self.create_draft(self.payload(expire_date=None)) response = self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") self.assertEqual(response.status_code, 400) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.DRAFT) self.assertEqual(InventoryBatch.objects.count(), 0) self.assertEqual(StockMovement.objects.count(), 0) def test_complete_existing_receipt_reconciles_draft_without_duplicate_stock(self): purchase = self.create_draft() batch = InventoryBatch.objects.create( product=self.product, batch_number="B-100", manufacture_date=timezone.localdate(), expiry_date=timezone.localdate() + timedelta(days=365), created_by=self.user, ) receive_stock( warehouse=self.warehouse, product=self.product, batch=batch, quantity="10", unit_cost="12.50", currency=self.currency, movement_type=StockMovement.MovementType.PURCHASE, reference_type="Purchase", reference_id=purchase.pk, created_by=self.user, ) response = self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") self.assertEqual(response.status_code, 200, response.data) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.POSTED) self.assertEqual(StockMovement.objects.count(), 1) self.assertEqual( StockBalance.objects.get(warehouse=self.warehouse).quantity, Decimal("10.0000"), ) def test_mismatched_existing_receipt_is_not_posted_or_duplicated(self): purchase = self.create_draft() batch = InventoryBatch.objects.create( product=self.product, batch_number="B-100", manufacture_date=timezone.localdate(), expiry_date=timezone.localdate() + timedelta(days=365), created_by=self.user, ) receive_stock( warehouse=self.warehouse, product=self.product, batch=batch, quantity="9", unit_cost="12.50", currency=self.currency, movement_type=StockMovement.MovementType.PURCHASE, reference_type="Purchase", reference_id=purchase.pk, created_by=self.user, ) response = self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") self.assertEqual(response.status_code, 400, response.data) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.DRAFT) self.assertEqual(StockMovement.objects.count(), 1) self.assertEqual( StockBalance.objects.get(warehouse=self.warehouse).quantity, Decimal("9.0000"), ) def test_draft_with_receipt_history_cannot_be_edited(self): purchase = self.create_draft() batch = InventoryBatch.objects.create( product=self.product, batch_number="B-100", manufacture_date=timezone.localdate(), expiry_date=timezone.localdate() + timedelta(days=365), created_by=self.user, ) receive_stock( warehouse=self.warehouse, product=self.product, batch=batch, quantity="10", unit_cost="12.50", currency=self.currency, movement_type=StockMovement.MovementType.PURCHASE, reference_type="Purchase", reference_id=purchase.pk, created_by=self.user, ) response = self.client.put( f"/api/shop/purchases/{purchase.pk}/", self.payload(batch_number="B-CHANGED"), format="json", ) self.assertEqual(response.status_code, 400, response.data) purchase.refresh_from_db() self.assertEqual(purchase.purchase_items.get().batch_number, "B-100") def test_draft_with_stock_ledger_history_cannot_be_deleted(self): purchase = self.create_draft() batch = InventoryBatch.objects.create( product=self.product, batch_number="B-100", manufacture_date=timezone.localdate(), expiry_date=timezone.localdate() + timedelta(days=365), created_by=self.user, ) receive_stock( warehouse=self.warehouse, product=self.product, batch=batch, quantity="10", unit_cost="12.50", currency=self.currency, movement_type=StockMovement.MovementType.PURCHASE, reference_type="Purchase", reference_id=purchase.pk, created_by=self.user, ) response = self.client.delete(f"/api/shop/purchases/{purchase.pk}/") self.assertEqual(response.status_code, 400, response.data) self.assertIn("cannot be deleted", str(response.data)) self.assertTrue(Purchase.objects.filter(pk=purchase.pk).exists()) self.assertEqual(StockMovement.objects.count(), 1) def test_posted_purchase_is_immutable_and_void_reverses_stock(self): purchase = self.create_draft() self.assertEqual( self.client.post(f"/api/shop/purchases/{purchase.pk}/post/").status_code, 200, ) edit = self.client.put( f"/api/shop/purchases/{purchase.pk}/", self.payload(), format="json" ) self.assertEqual(edit.status_code, 400) self.assertEqual( self.client.delete(f"/api/shop/purchases/{purchase.pk}/").status_code, 400, ) self.assertEqual( self.client.post( f"/api/shop/purchases/{purchase.pk}/void/", {}, format="json" ).status_code, 400, ) response = self.client.post( f"/api/shop/purchases/{purchase.pk}/void/", {"reason": "Supplier invoice cancelled"}, format="json", ) self.assertEqual(response.status_code, 200, response.data) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.VOID) self.assertEqual(StockBalance.objects.get(warehouse=self.warehouse).quantity, 0) self.assertEqual( StockMovement.objects.filter(movement_type="VOID_REVERSAL").count(), 1 ) def test_void_rolls_back_when_received_stock_was_consumed(self): purchase = self.create_draft() self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") movement = StockMovement.objects.get(movement_type="PURCHASE") issue_stock_batch( warehouse=self.warehouse, product=self.product, batch=movement.batch, quantity="1", movement_type=StockMovement.MovementType.SALE, ) response = self.client.post( f"/api/shop/purchases/{purchase.pk}/void/", {"reason": "Cannot keep it"}, format="json", ) self.assertEqual(response.status_code, 400) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.POSTED) self.assertEqual( StockMovement.objects.filter(movement_type="VOID_REVERSAL").count(), 0 ) self.assertEqual(StockBalance.objects.get(warehouse=self.warehouse).quantity, 9) def test_void_automatically_completes_recoverable_partial_reversal(self): purchase = self.create_draft() self.client.post(f"/api/shop/purchases/{purchase.pk}/post/") receipt = StockMovement.objects.get(movement_type="PURCHASE") issue_stock_batch( warehouse=receipt.warehouse, product=receipt.product, batch=receipt.batch, quantity="4", movement_type=StockMovement.MovementType.VOID_REVERSAL, reference_type="PurchaseVoid", reference_id=purchase.pk, remarks="Simulated interrupted cancellation", created_by=self.user, ) response = self.client.post( f"/api/shop/purchases/{purchase.pk}/void/", {"reason": "Complete cancellation"}, format="json", ) self.assertEqual(response.status_code, 200, response.data) purchase.refresh_from_db() self.assertEqual(purchase.document_status, Purchase.DocumentStatus.VOID) self.assertEqual(StockBalance.objects.get(warehouse=self.warehouse).quantity, 0) reversed_total = StockMovement.objects.filter( movement_type=StockMovement.MovementType.VOID_REVERSAL, reference_id=str(purchase.pk), ).aggregate(total=Sum("quantity_out"))["total"] self.assertEqual(reversed_total, Decimal("10.0000"))