from datetime import timedelta from decimal import Decimal from django.contrib.auth import get_user_model 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.sale import Sale, SaleItemBatchAllocation from shop_management.models.data.shop import Shop from shop_management.models.data.customer import Customer from shop_management.models.data.customer_payment import CustomerPayment, CustomerPaymentAllocation from shop_management.models.data.shop_account import Account from shop_management.models.data.utils import Currency, Unit from shop_management.services.inventory import receive_stock from shop_management.services.customer_payments import post_customer_payment class SalePostingTests(TestCase): def setUp(self): self.user = get_user_model().objects.create_user(username="seller", password="test", is_staff=True) self.user.allowed_tabs = ["sales"] 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.customer = Customer.objects.create(shop=self.shop, name="Registered Customer") self.account = Account.objects.create(shop=self.shop, name="Main Cash", currency=self.currency) self.unit = Unit.objects.create(shop=self.shop, title="Each") self.product = Product.objects.create( shop=self.shop, name="Medicine", default_unit=self.unit, requires_batch=True, requires_expiry=True, ) self.early = InventoryBatch.objects.create(product=self.product, batch_number="EARLY", expiry_date=timezone.localdate() + timedelta(days=30)) self.late = InventoryBatch.objects.create(product=self.product, batch_number="LATE", expiry_date=timezone.localdate() + timedelta(days=90)) self.expired = InventoryBatch.objects.create(product=self.product, batch_number="OLD", expiry_date=timezone.localdate() - timedelta(days=1)) receive_stock(warehouse=self.warehouse, product=self.product, batch=self.early, quantity=3, unit_cost=10, currency=self.currency) receive_stock(warehouse=self.warehouse, product=self.product, batch=self.late, quantity=5, unit_cost=20, currency=self.currency) receive_stock(warehouse=self.warehouse, product=self.product, batch=self.expired, quantity=10, unit_cost=1, currency=self.currency) StockMovement.objects.all().delete() self.client = APIClient() self.client.force_authenticate(self.user) def payload(self, quantity=5, allocations=None, sale_price="30"): item = {"product": self.product.id, "unit": self.unit.id, "quantity": quantity, "sale_price": sale_price} if allocations is not None: item["batch_allocations"] = allocations return { "shop": self.shop.id, "warehouse": self.warehouse.id, "currency": self.currency.id, "customer": self.customer.id, "sales_type": "CASH", "invoice_date": str(timezone.localdate()), "discount_type": "PERCENTAGE", "discount_value": "10", "received_amount": "0", "items": [item], } def create_draft(self, payload=None): response = self.client.post("/api/shop/sales/", payload or self.payload(), format="json") self.assertEqual(response.status_code, 201, response.data) return Sale.objects.get(pk=response.data["id"]) def test_fefo_post_calculates_server_totals_and_cogs(self): sale = self.create_draft() self.assertEqual(StockMovement.objects.count(), 0) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, Decimal("3")) self.assertEqual(StockBalance.objects.get(batch=self.late).quantity, Decimal("5")) self.assertEqual(self.customer.get_total_owed_by_currency(), {}) posted = self.client.post(f"/api/shop/sales/{sale.id}/post/") self.assertEqual(posted.status_code, 200, posted.data) sale.refresh_from_db() item = sale.sale_items.get() allocations = list(item.batch_allocations.order_by("batch__expiry_date")) self.assertEqual([row.quantity for row in allocations], [Decimal("3"), Decimal("2")]) self.assertEqual(item.cogs_total, Decimal("70.00")) self.assertEqual(sale.calculated_subtotal, Decimal("150.00")) self.assertEqual(sale.discount_amount, Decimal("15.00")) self.assertEqual(sale.calculated_total, Decimal("135.00")) self.assertEqual(self.customer.get_total_owed_by_currency(), {"AFN": 135.0}) self.assertRegex(sale.sale_number, r"^KAB-SAL-\d{4}-\d{6}$") self.assertEqual(StockBalance.objects.get(batch=self.expired).quantity, 10) def test_four_decimal_sale_price_is_preserved_through_posting(self): sale = self.create_draft(self.payload(quantity=2, sale_price="30.4040")) item = sale.sale_items.get() self.assertEqual(item.sale_price, Decimal("30.4040")) response = self.client.post(f"/api/shop/sales/{sale.id}/post/") self.assertEqual(response.status_code, 200, response.data) item.refresh_from_db() sale.refresh_from_db() self.assertEqual(item.sale_price, Decimal("30.4040")) self.assertEqual(sale.calculated_subtotal, Decimal("60.81")) def test_sale_edit_preserves_four_decimal_unit_price(self): sale = self.create_draft() response = self.client.patch( f"/api/shop/sales/{sale.id}/", self.payload(quantity=2, sale_price="30.4040"), format="json", ) self.assertEqual(response.status_code, 200, response.data) sale_item = sale.sale_items.get() self.assertEqual(sale_item.sale_price, Decimal("30.4040")) def test_complete_immediately_creates_posted_sale_in_one_request(self): payload = self.payload(quantity=2) payload["complete_immediately"] = True response = self.client.post("/api/shop/sales/", payload, format="json") self.assertEqual(response.status_code, 201, response.data) sale = Sale.objects.get(pk=response.data["id"]) self.assertEqual(sale.document_status, Sale.DocumentStatus.POSTED) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, Decimal("1")) self.assertEqual(self.customer.get_total_owed_by_currency(), {"AFN": 54.0}) def test_failed_immediate_completion_does_not_leave_a_draft(self): payload = self.payload(quantity=99) payload["complete_immediately"] = True response = self.client.post("/api/shop/sales/", payload, format="json") self.assertEqual(response.status_code, 400) self.assertEqual(Sale.objects.count(), 0) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, Decimal("3")) def test_manual_batch_override_and_void(self): sale = self.create_draft(self.payload(quantity=2, allocations=[{"batch": self.late.id, "quantity": "2"}])) self.client.post(f"/api/shop/sales/{sale.id}/post/") self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, 3) self.assertEqual(StockBalance.objects.get(batch=self.late).quantity, 3) voided = self.client.post(f"/api/shop/sales/{sale.id}/void/", {"reason": "Invoice cancelled"}, format="json") self.assertEqual(voided.status_code, 200, voided.data) self.assertEqual(StockBalance.objects.get(batch=self.late).quantity, 5) self.assertEqual(StockMovement.objects.filter(movement_type="VOID_REVERSAL").count(), 1) def test_insufficient_valid_stock_is_atomic(self): sale = self.create_draft(self.payload(quantity=9)) response = self.client.post(f"/api/shop/sales/{sale.id}/post/") self.assertEqual(response.status_code, 400) sale.refresh_from_db() self.assertEqual(sale.document_status, "DRAFT") self.assertEqual(StockMovement.objects.count(), 0) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, 3) def test_posted_sale_cannot_be_edited_deleted_or_reposted(self): sale = self.create_draft(self.payload(quantity=1)) self.client.post(f"/api/shop/sales/{sale.id}/post/") self.assertEqual(self.client.put(f"/api/shop/sales/{sale.id}/", self.payload(), format="json").status_code, 400) self.assertEqual(self.client.delete(f"/api/shop/sales/{sale.id}/").status_code, 400) self.assertEqual(self.client.post(f"/api/shop/sales/{sale.id}/post/").status_code, 400) def test_general_customer_cannot_be_posted_without_full_payment_account(self): payload = self.payload(quantity=1) payload["customer"] = "general-customer" sale = self.create_draft(payload) response = self.client.post(f"/api/shop/sales/{sale.id}/post/", {}, format="json") self.assertEqual(response.status_code, 400) self.assertIn("financial_account", response.data) sale.refresh_from_db() self.assertEqual(sale.document_status, Sale.DocumentStatus.DRAFT) def test_general_customer_post_creates_full_posted_receipt(self): payload = self.payload(quantity=1) payload["customer"] = "general-customer" sale = self.create_draft(payload) response = self.client.post( f"/api/shop/sales/{sale.id}/post/", {"financial_account": self.account.id}, format="json", ) self.assertEqual(response.status_code, 200, response.data) sale.refresh_from_db() payment = CustomerPayment.objects.get(allocations__sale=sale) self.assertEqual(payment.status, CustomerPayment.Status.POSTED) self.assertEqual(payment.amount, sale.calculated_total) self.assertEqual(payment.unallocated_amount, Decimal("0")) def test_named_customer_sale_with_direct_full_payment(self): payload = self.payload(quantity=1) payload["complete_immediately"] = True payload["received_amount"] = 27 payload["financial_account"] = self.account.id response = self.client.post("/api/shop/sales/", payload, format="json") self.assertEqual(response.status_code, 201, response.data) sale = Sale.objects.get(pk=response.data["id"]) self.assertEqual(sale.document_status, Sale.DocumentStatus.POSTED) payment = CustomerPayment.objects.get(allocations__sale=sale) self.assertEqual(payment.status, CustomerPayment.Status.POSTED) self.assertEqual(payment.amount, Decimal("27.00")) self.assertEqual(sale.total_paid[self.currency.title], Decimal("27.00")) self.assertEqual(sale.total_remained_amount[self.currency.title], Decimal("0.00")) def test_named_customer_sale_with_direct_partial_payment(self): payload = self.payload(quantity=1) payload["complete_immediately"] = True payload["received_amount"] = 10 payload["financial_account"] = self.account.id response = self.client.post("/api/shop/sales/", payload, format="json") self.assertEqual(response.status_code, 201, response.data) sale = Sale.objects.get(pk=response.data["id"]) self.assertEqual(sale.document_status, Sale.DocumentStatus.POSTED) payment = CustomerPayment.objects.get(allocations__sale=sale) self.assertEqual(payment.status, CustomerPayment.Status.POSTED) self.assertEqual(payment.amount, Decimal("10.00")) self.assertEqual(sale.total_paid[self.currency.title], Decimal("10.00")) self.assertEqual(sale.total_remained_amount[self.currency.title], Decimal("17.00")) def test_voiding_sale_with_direct_payment_cancels_linked_payment(self): payload = self.payload(quantity=1) payload["complete_immediately"] = True payload["received_amount"] = 50 payload["financial_account"] = self.account.id response = self.client.post("/api/shop/sales/", payload, format="json") self.assertEqual(response.status_code, 201, response.data) sale = Sale.objects.get(pk=response.data["id"]) payment = CustomerPayment.objects.get(allocations__sale=sale) self.assertEqual(payment.status, CustomerPayment.Status.POSTED) # Cancel (void) the sale void_resp = self.client.post(f"/api/shop/sales/{sale.id}/void/", {"reason": "Customer cancelled order"}, format="json") self.assertEqual(void_resp.status_code, 200, void_resp.data) sale.refresh_from_db() payment.refresh_from_db() self.assertEqual(sale.document_status, Sale.DocumentStatus.VOID) self.assertEqual(payment.status, CustomerPayment.Status.VOID) def test_voiding_sale_releases_separately_recorded_receipt_allocation(self): sale = self.create_draft(self.payload(quantity=1)) self.assertEqual(self.client.post(f"/api/shop/sales/{sale.id}/post/").status_code, 200) payment = CustomerPayment.objects.create( customer=self.customer, shop=self.shop, financial_account=self.account, currency=self.currency, amount=Decimal("10.00"), payment_date=timezone.localdate(), reference="EXTERNAL-RECEIPT", created_by=self.user, ) CustomerPaymentAllocation.objects.create( payment=payment, sale=sale, allocated_amount=Decimal("10.00"), created_by=self.user, ) post_customer_payment(payment.id, user=self.user) response = self.client.post( f"/api/shop/sales/{sale.id}/void/", {"reason": "Invoice cancelled"}, format="json", ) self.assertEqual(response.status_code, 200, response.data) payment.refresh_from_db() self.assertEqual(payment.status, CustomerPayment.Status.POSTED) self.assertFalse(payment.allocations.filter(sale=sale).exists()) self.assertEqual(payment.unallocated_amount, Decimal("10.00")) def test_sale_snapshots_customer_previous_balance(self): first = self.payload(quantity=1) # 30 - 10% = 27 owed first["complete_immediately"] = True response = self.client.post("/api/shop/sales/", first, format="json") self.assertEqual(response.status_code, 201, response.data) self.assertEqual(Decimal(str(response.data["previous_balance"])), Decimal("0.00")) second = self.payload(quantity=2) # 60 - 10% = 54 second["complete_immediately"] = True second["received_amount"] = 4 second["financial_account"] = self.account.id response = self.client.post("/api/shop/sales/", second, format="json") self.assertEqual(response.status_code, 201, response.data) sale = Sale.objects.get(pk=response.data["id"]) self.assertEqual(sale.previous_balance, Decimal("27.00")) self.assertEqual(sale.total_remained_amount[self.currency.title], Decimal("50.00")) self.assertEqual(self.customer.get_total_owed_by_currency(), {"AFN": 77.0}) def _admin_for_revision(self): self.user.is_superuser = True self.user.save(update_fields=["is_superuser"]) self.client.force_authenticate(self.user) def test_admin_revision_replaces_invoice_and_reconciles_stock_and_journals(self): from shop_management.models.data.accounting import JournalEntry sale = self.create_draft(self.payload(quantity=2)) self.assertEqual(self.client.post(f"/api/shop/sales/{sale.pk}/post/").status_code, 200) self._admin_for_revision() payload = self.payload(quantity=4) payload["items"][0]["sale_price"] = "40" payload["notes"] = "Corrected quantity and price" response = self.client.post(f"/api/shop/sales/{sale.pk}/revise/", payload, format="json") self.assertEqual(response.status_code, 200, response.data) replacement = Sale.objects.get(pk=response.data["id"]) sale.refresh_from_db() self.assertEqual(sale.document_status, "VOID") self.assertEqual(replacement.document_status, "POSTED") self.assertEqual(replacement.replaces_sale_id, sale.pk) self.assertNotEqual(replacement.sale_number, sale.sale_number) self.assertEqual(replacement.calculated_total, Decimal("144.00")) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, 0) self.assertEqual(StockBalance.objects.get(batch=self.late).quantity, 4) self.assertEqual(self.customer.get_total_owed_by_currency(), {"AFN": 144.0}) self.assertEqual(JournalEntry.objects.get(source_type="SALE", source_id=str(sale.pk)).status, "VOID") self.assertEqual(JournalEntry.objects.get(source_type="SALE", source_id=str(replacement.pk)).status, "POSTED") self.assertEqual(self.client.post(f"/api/shop/sales/{sale.pk}/revise/", payload, format="json").status_code, 400) def test_revision_requires_superadmin(self): sale = self.create_draft(self.payload(quantity=2)) self.client.post(f"/api/shop/sales/{sale.pk}/post/") response = self.client.post(f"/api/shop/sales/{sale.pk}/revise/", self.payload(), format="json") self.assertEqual(response.status_code, 403) sale.refresh_from_db() self.assertEqual(sale.document_status, "POSTED") self.assertEqual(Sale.objects.count(), 1) def test_failed_revision_rolls_back_original_stock_and_receipts(self): payload = self.payload(quantity=2) payload.update(complete_immediately=True, received_amount="20", financial_account=self.account.pk) sale = self.create_draft(payload) payment = CustomerPayment.objects.get(allocations__sale=sale) self._admin_for_revision() response = self.client.post(f"/api/shop/sales/{sale.pk}/revise/", self.payload(quantity=99), format="json") self.assertEqual(response.status_code, 400, response.data) sale.refresh_from_db() payment.refresh_from_db() self.assertEqual(sale.document_status, "POSTED") self.assertEqual(payment.status, "POSTED") self.assertEqual(Sale.objects.count(), 1) self.assertEqual(StockBalance.objects.get(batch=self.early).quantity, 1) self.assertTrue(payment.allocations.filter(sale=sale).exists()) def test_revision_keeps_receipt_and_applies_credit_without_duplicate_payment(self): payload = self.payload(quantity=2) payload.update(complete_immediately=True, received_amount="54", financial_account=self.account.pk) sale = self.create_draft(payload) payment = CustomerPayment.objects.get(allocations__sale=sale) self._admin_for_revision() response = self.client.post(f"/api/shop/sales/{sale.pk}/revise/", self.payload(quantity=3), format="json") self.assertEqual(response.status_code, 200, response.data) replacement = Sale.objects.get(pk=response.data["id"]) payment.refresh_from_db() self.assertEqual(payment.status, "POSTED") self.assertEqual(CustomerPayment.objects.filter(status="POSTED").count(), 1) self.assertEqual(replacement.total_paid["AFN"], Decimal("54.00")) self.assertEqual(replacement.total_remained_amount["AFN"], Decimal("27.00")) def test_general_customer_revision_does_not_collect_full_payment_twice(self): payload = self.payload(quantity=2) payload.update(customer="general-customer", complete_immediately=True, financial_account=self.account.pk) sale = self.create_draft(payload) self._admin_for_revision() correction = self.payload(quantity=1) correction["customer"] = sale.customer_id response = self.client.post(f"/api/shop/sales/{sale.pk}/revise/", correction, format="json") self.assertEqual(response.status_code, 200, response.data) replacement = Sale.objects.get(pk=response.data["id"]) self.assertEqual(CustomerPayment.objects.filter(status="POSTED").count(), 1) self.assertEqual(replacement.total_paid["AFN"], Decimal("27.00")) self.assertEqual(replacement.total_remained_amount["AFN"], Decimal("0.00"))