from datetime import datetime, time, timedelta from decimal import Decimal import json from django.test import TestCase from account.models import User from shop_management.models.data.product import Product from rest_framework.test import APIClient from shop_management.models.data.product import Category from shop_management.models.data.shop import Shop from shop_management.models.data.customer import Customer from shop_management.models.data.purchase import Purchase, PurchaseItem from shop_management.models.data.purchase_payment import PurchasePayment from shop_management.models.data.supplier import Supplier from shop_management.models.data.stock import Stock from shop_management.models.data.stock_transfer import StockTransfer from shop_management.models.data.sale import Sale, SaleItem from shop_management.models.data.sale_return import SaleReturn from shop_management.models.data.shop_account import Account from shop_management.models.data.utils import Currency, Unit from shop_management.models.data.expense import Expense from shop_management.models.data.customer_debtor import CustomerDebtor from shop_management.models.data.passive_income import PassiveIncome from shop_management.models.data.account_transfer import AccountTransfer from shop_management.models.data.sale_payment import SalePayment from shop_management.models.data.customer_balance_payment import CustomerBalancePayment from django.utils import timezone class ProductModelSchemaTests(TestCase): def test_product_model_uses_code_and_not_removed_fields(self): field_names = {field.name for field in Product._meta.get_fields()} self.assertIn("code", field_names) self.assertNotIn("pack", field_names) self.assertNotIn("carton", field_names) self.assertNotIn("raf", field_names) class ProductApiTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="productstaff", password="testpass123", is_staff=True, allowed_tabs=["products", "category"], ) self.shop = Shop.objects.create( name="Test Shop", owner=self.user, ) self.category = Category.objects.create( name="Medicines", shop=self.shop, created_by=self.user, ) Product.objects.create( name="Paracetamol", code="P-100", shop=self.shop, category=self.category, created_by=self.user, ) self.branch_user = User.objects.create_user( username="branchstaff", password="testpass123", is_staff=True, allowed_tabs=["products", "sales", "sale_return"], ) self.shop.assigned_users.add(self.branch_user) self.currency = Currency.objects.create( title="AFG", shop=self.shop, created_by=self.user, ) self.unit = Unit.objects.create( title="PCS", shop=self.shop, created_by=self.user, ) self.customer = Customer.objects.create( name="Branch Customer", shop=self.shop, created_by=self.user, ) self.purchase = Purchase.objects.create( shop=self.shop, currency=self.currency, created_by=self.user, ) self.purchase_item = PurchaseItem.objects.create( purchase=self.purchase, product=Product.objects.get(name="Paracetamol"), unit=self.unit, quantity=10, buy_price=10, retail_price=15, sale_price=15, created_by=self.user, ) self.stock = Stock.objects.create( shop=self.shop, purchase_item=self.purchase_item, quantity=10, sale_price=15, retail_price=15, created_by=self.user, ) def test_products_list_endpoint_returns_200(self): self.client.force_authenticate(user=self.user) response = self.client.get( "/api/shop/products/", {"search": "", "page": 1, "page_size": 10}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 1) def test_assigned_branch_user_can_create_sale_from_owner_stock(self): self.client.force_authenticate(user=self.branch_user) response = self.client.post( "/api/shop/sales/", { "shop": self.shop.id, "customer": self.customer.id, "sale_date": "2026-05-21", "currency": self.currency.id, "received_amount": "0", "amount": "0", "items": [ { "product": self.purchase_item.product_id, "purchase_item": self.purchase_item.id, "unit": self.unit.id, "quantity": 2, "sale_price": "15", } ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 201) self.stock.refresh_from_db() self.assertEqual(self.stock.quantity, 8) def test_assigned_branch_user_can_list_branch_products(self): self.client.force_authenticate(user=self.branch_user) response = self.client.get( "/api/shop/products/", {"search": "", "page": 1, "page_size": 10}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 1) def test_sale_return_cannot_add_more_than_was_sold(self): self.client.force_authenticate(user=self.branch_user) sale_response = self.client.post( "/api/shop/sales/", { "shop": self.shop.id, "customer": self.customer.id, "sale_date": "2026-05-21", "currency": self.currency.id, "received_amount": "0", "amount": "0", "items": [{ "product": self.purchase_item.product_id, "purchase_item": self.purchase_item.id, "unit": self.unit.id, "quantity": 2, "sale_price": "15", }], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(sale_response.status_code, 201, sale_response.data) sale_item = SaleItem.objects.get(sale__shop=self.shop) self.stock.refresh_from_db() self.assertEqual(self.stock.quantity, 8) invalid_return = self.client.post( "/api/shop/sale-returns/", { "shop": self.shop.id, "sale": sale_item.sale_id, "return_items": [ { "sale_item": sale_item.id, "return_quantity": 1, "refund_amount": "15", }, { "sale_item": sale_item.id, "return_quantity": 2, "refund_amount": "30", }, ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(invalid_return.status_code, 400, invalid_return.data) self.stock.refresh_from_db() self.assertEqual(self.stock.quantity, 8) self.assertFalse(SaleReturn.objects.filter(sale_id=sale_item.sale_id).exists()) valid_return = self.client.post( "/api/shop/sale-returns/", { "shop": self.shop.id, "sale": sale_item.sale_id, "return_items": [{ "sale_item": sale_item.id, "return_quantity": 2, "refund_amount": "30", }], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(valid_return.status_code, 201, valid_return.data) self.stock.refresh_from_db() self.assertEqual(self.stock.quantity, 10) delete_sale = self.client.delete( f"/api/shop/sales/{sale_item.sale_id}/", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(delete_sale.status_code, 204, delete_sale.data) self.stock.refresh_from_db() self.assertEqual(self.stock.quantity, 10) self.assertFalse(SaleReturn.objects.filter(sale_id=sale_item.sale_id).exists()) class StockImportExcelTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="stockimport", password="testpass123", is_staff=True, allowed_tabs=["stock"], ) self.shop = Shop.objects.create(name="Main", owner=self.user) self.client.force_authenticate(user=self.user) def test_import_excel_keeps_afg_and_usd_price_columns(self): response = self.client.post( "/api/shop/stocks/import_excel/", { "shop": self.shop.id, "rows": [ { "product": "DX 2018 SCREEN", "code": "101", "buy_price_afg": "13500", "buy_price_usd": "200", "wholesale_price_afg": "16000", "wholesale_price_usd": "210", "retail_price_afg": "17500", "retail_price_usd": "220", "available_qty": "16", "unit": "PCS", "branch": "Main", } ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 201, response.data) stock = Stock.objects.get(purchase_item__product__code="101") price_by_currency = json.loads(stock.comments)["price_by_currency"] self.assertEqual(price_by_currency["buy_price_afg"], "13500.0") self.assertEqual(price_by_currency["buy_price_usd"], "200.0") class AccountApiTests(TestCase): def setUp(self): self.client = APIClient() self.owner = User.objects.create_user( username="accountowner", password="testpass123", is_staff=True, allowed_tabs=["accounts", "currencies"], ) self.branch_user = User.objects.create_user( username="accountbranch", password="testpass123", is_staff=True, allowed_tabs=["accounts", "currencies"], ) self.shop = Shop.objects.create(name="Basit Branch", owner=self.owner) self.shop.assigned_users.add(self.branch_user) self.currency = Currency.objects.create( title="USD", shop=self.shop, created_by=self.owner, ) self.account = Account.objects.create( name="Main USD ACCOUNT", shop=self.shop, currency=self.currency, opening_balance="100.00", ) def test_assigned_branch_user_can_list_branch_accounts(self): self.client.force_authenticate(user=self.branch_user) response = self.client.get( "/api/shop/accounts/", {"page": 1, "page_size": 10}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 1) self.assertEqual(response.data["results"][0]["id"], self.account.id) def test_accounts_can_be_filtered_by_active_state(self): inactive = Account.objects.create( name="Old USD Account", shop=self.shop, currency=self.currency, is_active=False, ) self.client.force_authenticate(user=self.owner) response = self.client.get( "/api/shop/accounts/", {"page_size": "all", "is_active": "false"}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200) rows = response.data.get("results", response.data) self.assertEqual([row["id"] for row in rows], [inactive.id]) def test_account_with_opening_balance_cannot_be_deleted(self): self.client.force_authenticate(user=self.owner) response = self.client.delete( f"/api/shop/accounts/{self.account.id}/", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 400) self.assertTrue(Account.objects.filter(pk=self.account.id).exists()) self.assertIn("Deactivate it instead", response.data["detail"]) def test_unused_zero_balance_account_can_be_deleted(self): unused = Account.objects.create( name="Mistake Account", shop=self.shop, currency=self.currency, ) self.client.force_authenticate(user=self.owner) response = self.client.delete( f"/api/shop/accounts/{unused.id}/", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 204) self.assertFalse(Account.objects.filter(pk=unused.id).exists()) class PurchaseWeightedAverageCostTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="purchasewac", password="testpass123", is_staff=True, allowed_tabs=[ "purchases", "purchase_return", "products", "stock", "units", "currencies", "category", "suppliers", ], ) self.shop = Shop.objects.create(name="WAC Shop", owner=self.user) self.category = Category.objects.create( name="General", shop=self.shop, created_by=self.user, ) self.unit = Unit.objects.create( title="PCS", shop=self.shop, created_by=self.user, ) self.product = Product.objects.create( name="Cup", code="CUP-1", shop=self.shop, category=self.category, default_unit=self.unit, created_by=self.user, ) self.currency = Currency.objects.create( title="AFG", shop=self.shop, created_by=self.user, ) self.client.force_authenticate(user=self.user) def _create_purchase(self, quantity, buy_price): response = self.client.post( "/api/shop/purchases/", { "shop": self.shop.id, "supplier": "", "currency": self.currency.id, "amount": "0", "received_amount": "0", "purchase_date": "2026-06-29", "items": [ { "product": self.product.id, "unit": self.unit.id, "quantity": str(quantity), "buy_price": str(buy_price), "wholesale_price": str(Decimal(buy_price) + Decimal("2")), "retail_price": str(Decimal(buy_price) + Decimal("4")), "sale_price": str(Decimal(buy_price) + Decimal("4")), } ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 201, response.data) def test_purchase_creates_weighted_average_cost_stock(self): self._create_purchase(100, Decimal("10")) self._create_purchase(160, Decimal("15")) stock = Stock.objects.get(shop=self.shop, purchase_item__product=self.product) stock.refresh_from_db() self.assertEqual(stock.quantity, 260) self.assertEqual(stock.average_buy_price.quantize(Decimal("0.01")), Decimal("13.08")) self.assertEqual(stock.average_wholesale_price.quantize(Decimal("0.01")), Decimal("15.08")) self.assertEqual(stock.average_retail_price.quantize(Decimal("0.01")), Decimal("17.08")) def test_purchase_return_deducts_once_and_delete_restores_once(self): self._create_purchase(100, Decimal("10")) purchase_item = PurchaseItem.objects.get(product=self.product) stock = Stock.objects.get(shop=self.shop, purchase_item__product=self.product) create_return = self.client.post( "/api/shop/purchase-returns/", { "purchase": purchase_item.purchase_id, "shop": self.shop.id, "reason": "Supplier return", "return_items": [{ "purchase_item": purchase_item.id, "return_quantity": 10, "refund_amount": "100", }], }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(create_return.status_code, 201, create_return.data) stock.refresh_from_db() self.assertEqual(stock.quantity, 90) delete_return = self.client.delete( f"/api/shop/purchase-returns/{create_return.data['id']}/", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(delete_return.status_code, 204, delete_return.data) stock.refresh_from_db() self.assertEqual(stock.quantity, 100) class FinancialDashboardPurchaseTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="dashboardstaff", password="testpass123", is_staff=True, allowed_tabs=["dashboard", "purchases", "products", "units", "currencies", "category"], ) self.shop = Shop.objects.create(name="Dashboard Shop", owner=self.user) self.category = Category.objects.create( name="General", shop=self.shop, created_by=self.user, ) self.product = Product.objects.create( name="Bottle", code="BOT-1", shop=self.shop, category=self.category, created_by=self.user, ) self.unit = Unit.objects.create( title="PCS", shop=self.shop, created_by=self.user, ) self.currency = Currency.objects.create( title="AFG", shop=self.shop, created_by=self.user, ) self.client.force_authenticate(user=self.user) @staticmethod def _noon_cycle_date(when=None): local = timezone.localtime(when) if when is not None else timezone.localtime() day = local.date() if local.time() < time(12, 0): return day - timedelta(days=1) return day def _daily_dashboard_query(self, day=None, shop=True): day = day or self._noon_cycle_date() params = { "start_date": day.isoformat(), "end_date": day.isoformat(), "day_cycle": "noon", } if shop: params["shop"] = self.shop.id return params def test_dashboard_includes_supplier_purchase_totals(self): supplier = Supplier.objects.create( name="Main Supplier", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=supplier, currency=self.currency, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), retail_price=Decimal("120"), sale_price=Decimal("120"), created_by=self.user, ) PurchasePayment.objects.create( purchase=purchase, currency=self.currency, amount=Decimal("1500"), payment_date=timezone.now(), converted_amount=Decimal("1500"), exchange_rate=Decimal("1"), created_by=self.user, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_purchases"], [{"currency": "AFG", "amount": "1000.00"}], ) self.assertEqual( response.data["total_purchases_paid"], [{"currency": "AFG", "amount": "1000.00"}], ) self.assertEqual( response.data["total_purchases_unpaid"], [], ) def test_dashboard_shows_supplierless_usd_purchase_as_paid_only(self): usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=None, currency=usd, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), retail_price=Decimal("120"), sale_price=Decimal("120"), created_by=self.user, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_purchases"], [{"currency": "USD", "amount": "1000.00"}], ) self.assertEqual( response.data["total_purchases_paid"], [{"currency": "USD", "amount": "1000.00"}], ) self.assertEqual(response.data["total_purchases_unpaid"], []) def test_dashboard_shows_usd_purchase_paid_and_unpaid_by_invoice_currency(self): usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user, ) supplier = Supplier.objects.create( name="Multi Currency Supplier", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=supplier, currency=usd, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), retail_price=Decimal("120"), sale_price=Decimal("120"), created_by=self.user, ) PurchasePayment.objects.create( purchase=purchase, currency=self.currency, amount=Decimal("500"), payment_date=timezone.now(), converted_amount=Decimal("500"), exchange_rate=Decimal("1"), created_by=self.user, ) PurchasePayment.objects.create( purchase=purchase, currency=usd, amount=Decimal("200"), payment_date=timezone.now(), converted_amount=Decimal("200"), exchange_rate=Decimal("1"), created_by=self.user, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_purchases"], [{"currency": "USD", "amount": "1000.00"}], ) self.assertEqual( response.data["total_purchases_paid"], [{"currency": "USD", "amount": "700.00"}], ) self.assertEqual( response.data["total_purchases_unpaid"], [{"currency": "USD", "amount": "300.00"}], ) def test_dashboard_and_purchase_totals_keep_mixed_currencies_separate(self): usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user, ) supplier = Supplier.objects.create( name="Transfer Supplier", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=supplier, currency=self.currency, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, currency=usd, product=self.product, unit=self.unit, quantity=2, buy_price=Decimal("30"), retail_price=Decimal("50"), sale_price=Decimal("50"), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=1, buy_price=Decimal("100"), retail_price=Decimal("120"), sale_price=Decimal("120"), created_by=self.user, ) self.assertEqual( purchase.total_price, {"USD": Decimal("60"), "AFG": Decimal("100")}, ) self.assertEqual( purchase.total_remained_amount, {"USD": Decimal("60"), "AFG": Decimal("100")}, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertCountEqual( response.data["total_purchases"], [ {"currency": "USD", "amount": "60.00"}, {"currency": "AFG", "amount": "100.00"}, ], ) self.assertCountEqual( response.data["total_purchases_unpaid"], [ {"currency": "USD", "amount": "60.00"}, {"currency": "AFG", "amount": "100.00"}, ], ) def test_mixed_currency_transfer_payment_clears_dashboard_unpaid(self): usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user, ) supplier = Supplier.objects.create( name="Transfer from Kabul Main Branch", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=supplier, currency=self.currency, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, currency=usd, product=self.product, unit=self.unit, quantity=1, buy_price=Decimal("1400"), retail_price=Decimal("1500"), sale_price=Decimal("1500"), created_by=self.user, ) PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=1, buy_price=Decimal("12000"), retail_price=Decimal("13000"), sale_price=Decimal("13000"), created_by=self.user, ) # Paid the combined invoice in AFG only (how Takhar transfer payments were stored). PurchasePayment.objects.create( purchase=purchase, currency=self.currency, amount=Decimal("13400"), payment_date=timezone.now(), converted_amount=Decimal("13400"), exchange_rate=Decimal("1"), created_by=self.user, ) self.assertEqual( purchase.total_price, {"USD": Decimal("1400"), "AFG": Decimal("12000")}, ) self.assertEqual( purchase.total_remained_amount, {"USD": Decimal("0"), "AFG": Decimal("0")}, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual(response.data["total_purchases_unpaid"], []) self.assertCountEqual( response.data["total_purchases_paid"], [ {"currency": "USD", "amount": "1400.00"}, {"currency": "AFG", "amount": "12000.00"}, ], ) def test_payment_off_transfer_purchase_has_no_unpaid_when_links_are_broken(self): source_shop = Shop.objects.create(name="Kabul Main Branch", owner=self.user) supplier = Supplier.objects.create( name="Transfer from Kabul Main Branch", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, supplier=supplier, currency=self.currency, amount=Decimal("0"), received_amount=Decimal("0"), purchase_date=timezone.localdate(), created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), retail_price=Decimal("120"), sale_price=Decimal("120"), created_by=self.user, ) source_purchase = Purchase.objects.create( shop=source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) source_item = PurchaseItem.objects.create( purchase=source_purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), created_by=self.user, ) source_stock = Stock.objects.create( shop=source_shop, purchase_item=source_item, quantity=10, sale_price=Decimal("100"), average_buy_price=Decimal("100"), created_by=self.user, ) # Merged destination stock points at a different purchase, breaking the link. orphan_purchase = Purchase.objects.create( shop=self.shop, supplier=None, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) orphan_item = PurchaseItem.objects.create( purchase=orphan_purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("100"), created_by=self.user, ) target_stock = Stock.objects.create( shop=self.shop, purchase_item=orphan_item, quantity=10, sale_price=Decimal("100"), average_buy_price=Decimal("100"), created_by=self.user, ) StockTransfer.objects.create( transfer_id="TRF-UNPAID-1", product=self.product, source_stock=source_stock, target_stock=target_stock, supplier_branch=source_shop, customer_branch=self.shop, currency=self.currency, transfer_quantity=10, sale_price=Decimal("100"), old_quantity=20, new_quantity=10, transfer_date=purchase.purchase_date, status=StockTransfer.Status.ACCEPTED, show_payment_button=False, created_by=self.user, ) self.assertEqual( purchase.total_remained_amount, {"AFG": Decimal("0")}, ) response = self.client.get( "/api/shop/financial-report/", {"shop": self.shop.id}, HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) unpaid = response.data["total_purchases_unpaid"] self.assertEqual(unpaid, []) def test_daily_dashboard_includes_expense_created_today(self): account = Account.objects.create( shop=self.shop, name="Cash", currency=self.currency, opening_balance=Decimal("0"), ) Expense.objects.create( shop=self.shop, account=account, expense_by="Shop Manager", amount=Decimal("250.00"), currency=self.currency, payment_date=timezone.now(), created_by=self.user, ) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "250.00"}], ) def test_daily_dashboard_includes_midnight_local_expense(self): today = timezone.localdate() midnight_local = timezone.make_aware(datetime.combine(today, time.min)) account = Account.objects.create( shop=self.shop, name="Cash Midnight", currency=self.currency, opening_balance=Decimal("0"), ) Expense.objects.create( shop=self.shop, account=account, expense_by="Night Shift", amount=Decimal("75.50"), currency=self.currency, payment_date=midnight_local, created_by=self.user, ) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(self._noon_cycle_date(midnight_local)), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "75.50"}], ) def test_daily_dashboard_uses_noon_to_noon_window(self): today = timezone.localdate() yesterday = today - timedelta(days=1) tomorrow = today + timedelta(days=1) tz = timezone.get_current_timezone() account = Account.objects.create( shop=self.shop, name="Noon Cycle Cash", currency=self.currency, opening_balance=Decimal("0"), ) def make_expense(when, amount, name): Expense.objects.create( shop=self.shop, account=account, expense_by=name, amount=Decimal(amount), currency=self.currency, payment_date=when, created_by=self.user, ) make_expense(timezone.make_aware(datetime.combine(today, time(11, 59)), tz), "10.00", "Before noon") make_expense(timezone.make_aware(datetime.combine(today, time(12, 0)), tz), "20.00", "Noon start") make_expense(timezone.make_aware(datetime.combine(today, time(22, 0)), tz), "30.00", "Evening") make_expense(timezone.make_aware(datetime.combine(tomorrow, time(1, 0)), tz), "40.00", "After midnight") make_expense(timezone.make_aware(datetime.combine(tomorrow, time(12, 0)), tz), "50.00", "Next noon") today_response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(today), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(today_response.status_code, 200, today_response.data) self.assertEqual( today_response.data["total_expenses"], [{"currency": "AFG", "amount": "90.00"}], ) yesterday_response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(yesterday), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(yesterday_response.status_code, 200, yesterday_response.data) self.assertEqual( yesterday_response.data["total_expenses"], [{"currency": "AFG", "amount": "10.00"}], ) def test_daily_dashboard_includes_expense_for_assigned_branch_user(self): branch_user = User.objects.create_user( username="dashboardbranch", password="testpass123", is_staff=True, allowed_tabs=["dashboard", "expenses", "accounts"], ) self.shop.assigned_users.add(branch_user) account = Account.objects.create( shop=self.shop, name="Branch Cash", currency=self.currency, opening_balance=Decimal("0"), ) Expense.objects.create( shop=self.shop, account=account, expense_by="Branch Staff", amount=Decimal("40.00"), currency=self.currency, payment_date=timezone.now(), created_by=branch_user, ) self.client.force_authenticate(user=branch_user) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "40.00"}], ) def test_daily_dashboard_includes_expense_created_via_api(self): self.user.allowed_tabs = list( {*self.user.allowed_tabs, "expenses", "accounts"} ) self.user.save(update_fields=["allowed_tabs"]) account = Account.objects.create( shop=self.shop, name="API Cash", currency=self.currency, opening_balance=Decimal("0"), ) business_day = self._noon_cycle_date() payment_date = f"{business_day.isoformat()}T14:15" create_response = self.client.post( "/api/shop/expenses/", { "shop": self.shop.id, "expense_by": "API User", "amount": "120.00", "currency": self.currency.id, "account": account.id, "exchange_rate": "1", "payment_date": payment_date, }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(create_response.status_code, 201, create_response.data) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(business_day), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "120.00"}], ) def test_daily_dashboard_includes_evening_local_expense(self): self.user.allowed_tabs = list( {*self.user.allowed_tabs, "expenses", "accounts"} ) self.user.save(update_fields=["allowed_tabs"]) account = Account.objects.create( shop=self.shop, name="Evening Cash", currency=self.currency, opening_balance=Decimal("0"), ) business_day = self._noon_cycle_date() create_response = self.client.post( "/api/shop/expenses/", { "shop": self.shop.id, "expense_by": "Night Shift", "amount": "80.00", "currency": self.currency.id, "account": account.id, "exchange_rate": "1", "payment_date": f"{business_day.isoformat()}T22:14", }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(create_response.status_code, 201, create_response.data) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(business_day), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "80.00"}], ) def test_daily_dashboard_includes_expense_for_superuser(self): superuser = User.objects.create_user( username="dashboardadmin", password="testpass123", is_staff=True, is_superuser=True, ) account = Account.objects.create( shop=self.shop, name="Admin Cash", currency=self.currency, opening_balance=Decimal("0"), ) Expense.objects.create( shop=self.shop, account=account, expense_by="Admin", amount=Decimal("99.00"), currency=self.currency, payment_date=timezone.now(), created_by=superuser, ) self.client.force_authenticate(user=superuser) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "99.00"}], ) all_branches = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(shop=False), ) self.assertEqual(all_branches.status_code, 200, all_branches.data) self.assertEqual( all_branches.data["total_expenses"], [{"currency": "AFG", "amount": "99.00"}], ) def test_daily_dashboard_expense_includes_debtors_and_transfer_out(self): today = self._noon_cycle_date() customer = Customer.objects.create( name="Debtor Customer", shop=self.shop, created_by=self.user, ) account = Account.objects.create( shop=self.shop, name="Main Cash", currency=self.currency, opening_balance=Decimal("0"), ) to_account = Account.objects.create( shop=self.shop, name="Bank Cash", currency=self.currency, opening_balance=Decimal("0"), ) Expense.objects.create( shop=self.shop, account=account, expense_by="Ops", amount=Decimal("100.00"), currency=self.currency, payment_date=timezone.now(), created_by=self.user, ) CustomerDebtor.objects.create( customer=customer, account=account, amount=Decimal("50.00"), currency=self.currency, loan_date=timezone.now(), created_by=self.user, ) AccountTransfer.objects.create( from_account=account, to_account=to_account, from_currency=self.currency, to_currency=self.currency, amount=Decimal("25.00"), exchange_rate=Decimal("1"), converted_amount=Decimal("25.00"), transfer_date=today, ) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(today), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_expenses"], [{"currency": "AFG", "amount": "175.00"}], ) def test_daily_dashboard_income_excludes_invoice_totals(self): today = self._noon_cycle_date() customer = Customer.objects.create( name="Income Customer", shop=self.shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.shop, currency=self.currency, purchase_date=today, created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("60"), retail_price=Decimal("100"), sale_price=Decimal("100"), created_by=self.user, ) account = Account.objects.create( shop=self.shop, name="Income Cash", currency=self.currency, opening_balance=Decimal("0"), ) sale = Sale.objects.create( shop=self.shop, customer=customer, currency=self.currency, sale_date=today, created_by=self.user, ) SaleItem.objects.create( sale=sale, product=self.product, purchase_item=purchase_item, unit=self.unit, quantity=2, sale_price=Decimal("100"), buy_price=Decimal("60"), created_by=self.user, ) PassiveIncome.objects.create( shop=self.shop, account=account, name="Rental", amount=Decimal("30.00"), currency=self.currency, exchange_rate=Decimal("1"), converted_amount=Decimal("30.00"), income_date=timezone.now(), created_by=self.user, ) SalePayment.objects.create( sale=sale, currency=self.currency, amount=Decimal("40.00"), payment_date=timezone.now(), created_by=self.user, ) CustomerBalancePayment.objects.create( customer=customer, account=account, amount=Decimal("15.00"), currency=self.currency, exchange_rate=Decimal("1"), converted_amount=Decimal("15.00"), payment_date=timezone.now(), created_by=self.user, ) transfer_from = Account.objects.create( shop=self.shop, name="Transfer Source", currency=self.currency, opening_balance=Decimal("0"), ) transfer_to = Account.objects.create( shop=self.shop, name="Transfer Target", currency=self.currency, opening_balance=Decimal("0"), ) AccountTransfer.objects.create( from_account=transfer_from, to_account=transfer_to, from_currency=self.currency, to_currency=self.currency, amount=Decimal("20.00"), exchange_rate=Decimal("1"), converted_amount=Decimal("20.00"), transfer_date=today, ) response = self.client.get( "/api/shop/financial-report/", self._daily_dashboard_query(today), HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual( response.data["total_sales"], [{"currency": "AFG", "amount": "200.00"}], ) self.assertEqual( response.data["total_collected"], [{"currency": "AFG", "amount": "15.00"}], ) self.assertEqual( response.data["total_incomes"], [{"currency": "AFG", "amount": "85.00"}], ) class StockTransferEditTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="transferedit", password="testpass123", is_staff=True, allowed_tabs=["stock", "purchases", "products", "units", "currencies", "category"], ) self.source_shop = Shop.objects.create(name="Source Branch", owner=self.user) self.target_shop = Shop.objects.create(name="Target Branch", owner=self.user) self.category = Category.objects.create( name="General", shop=self.source_shop, created_by=self.user, ) self.product = Product.objects.create( name="Engine Oil", code="ENG-1", shop=self.source_shop, category=self.category, created_by=self.user, ) self.unit = Unit.objects.create( title="PCS", shop=self.source_shop, created_by=self.user, ) self.currency = Currency.objects.create( title="USD", shop=self.source_shop, created_by=self.user, ) self.source_purchase = Purchase.objects.create( shop=self.source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) self.source_purchase_item = PurchaseItem.objects.create( purchase=self.source_purchase, product=self.product, unit=self.unit, quantity=10, buy_price=Decimal("10"), wholesale_price=Decimal("15"), retail_price=Decimal("20"), sale_price=Decimal("20"), created_by=self.user, ) self.source_stock = Stock.objects.create( shop=self.source_shop, purchase_item=self.source_purchase_item, quantity=10, sale_price=Decimal("20"), average_buy_price=Decimal("10"), wholesale_price=Decimal("15"), retail_price=Decimal("20"), created_by=self.user, ) self.transfer = StockTransfer.objects.create( transfer_id="TRF-TEST-001", status=StockTransfer.Status.PENDING, product=self.product, source_stock=self.source_stock, supplier_branch=self.source_shop, customer_branch=self.target_shop, currency=self.currency, transfer_quantity=4, sale_price=Decimal("20"), old_quantity=10, new_quantity=6, source_quantity_reserved=True, transfer_date=timezone.localdate(), created_by=self.user, ) self.source_stock.quantity = 6 self.source_stock.save(update_fields=["quantity", "updated_at"]) self.client.force_authenticate(user=self.user) def test_edit_accepted_transfer_updates_both_source_and_target_stock(self): accept_response = self.client.post( f"/api/shop/stock-transfers/{self.transfer.id}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(accept_response.status_code, 200, accept_response.data) self.source_stock.refresh_from_db() self.transfer.refresh_from_db() target_stock = self.transfer.target_stock self.assertIsNotNone(target_stock) target_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 6) self.assertEqual(target_stock.quantity, 4) edit_response = self.client.patch( f"/api/shop/stock-transfers/{self.transfer.id}/", { "transfer_quantity": 6, "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(edit_response.status_code, 200, edit_response.data) self.source_stock.refresh_from_db() self.transfer.refresh_from_db() target_stock.refresh_from_db() target_purchase_item = target_stock.purchase_item target_purchase_item.refresh_from_db() self.assertEqual(self.source_stock.quantity, 4) self.assertEqual(target_stock.quantity, 6) self.assertEqual(target_purchase_item.quantity, 6) self.assertEqual(self.transfer.old_quantity, 10) self.assertEqual(self.transfer.new_quantity, 4) def test_accept_does_not_deduct_reserved_quantity_twice_after_stock_changes(self): # The transfer already reserved 4 (10 -> 6). A later purchase/adjustment # must not make acceptance deduct the same 4 again. self.source_stock.quantity = 11 self.source_stock.save(update_fields=["quantity", "updated_at"]) accept_response = self.client.post( f"/api/shop/stock-transfers/{self.transfer.id}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, accept_response.data) self.source_stock.refresh_from_db() self.transfer.refresh_from_db() self.assertEqual(self.source_stock.quantity, 11) self.assertEqual(self.transfer.target_stock.quantity, 4) self.assertFalse(self.transfer.source_quantity_reserved) def test_delete_pending_transfer_marks_it_rejected(self): delete_response = self.client.delete( f"/api/shop/stock-transfers/{self.transfer.id}/", format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(delete_response.status_code, 204, delete_response.data) self.transfer.refresh_from_db() self.assertEqual(self.transfer.status, StockTransfer.Status.REJECTED) self.assertIsNone(self.transfer.accepted_date) self.assertIsNone(self.transfer.accepted_datetime) self.assertIsNone(self.transfer.accepted_by) self.assertIsNone(self.transfer.target_stock) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 10) def test_delete_accepted_transfer_reverses_stock_and_removes_transfer(self): accept_response = self.client.post( f"/api/shop/stock-transfers/{self.transfer.id}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(accept_response.status_code, 200, accept_response.data) self.source_stock.refresh_from_db() self.transfer.refresh_from_db() target_stock = self.transfer.target_stock self.assertIsNotNone(target_stock) target_stock.refresh_from_db() target_stock_id = target_stock.id delete_response = self.client.delete( f"/api/shop/stock-transfers/{self.transfer.id}/", format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(delete_response.status_code, 204, delete_response.data) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 10) self.assertFalse(StockTransfer.objects.filter(id=self.transfer.id).exists()) self.assertFalse(Stock.objects.filter(id=target_stock_id).exists()) self.assertFalse(Purchase.objects.filter(shop=self.target_shop).exists()) def test_delete_accepted_transfer_is_blocked_after_destination_stock_is_used(self): accept_response = self.client.post( f"/api/shop/stock-transfers/{self.transfer.id}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, accept_response.data) self.transfer.refresh_from_db() target_stock = self.transfer.target_stock target_stock.quantity = 2 target_stock.save(update_fields=["quantity", "updated_at"]) delete_response = self.client.delete( f"/api/shop/stock-transfers/{self.transfer.id}/", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(delete_response.status_code, 400, delete_response.data) self.source_stock.refresh_from_db() target_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 6) self.assertEqual(target_stock.quantity, 2) self.assertTrue(StockTransfer.objects.filter(id=self.transfer.id).exists()) class StockTransferCrossShopCurrencyTests(TestCase): """Transfers must value destination stock with the destination shop currency.""" def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="transfercurrency", password="testpass123", is_staff=True, allowed_tabs=[ "stock", "stock_transfer", "sales", "purchases", "products", "units", "currencies", "category", ], ) self.source_shop = Shop.objects.create(name="Kabul Main Branch", owner=self.user) self.target_shop = Shop.objects.create(name="Takhar Branch", owner=self.user) self.category = Category.objects.create( name="General", shop=self.source_shop, created_by=self.user, ) self.product = Product.objects.create( name="SPARAKET BOLTS", code="SPB-1", shop=self.source_shop, category=self.category, created_by=self.user, ) self.unit = Unit.objects.create( title="PCS", shop=self.source_shop, created_by=self.user, ) self.source_usd = Currency.objects.create( title="USD", shop=self.source_shop, created_by=self.user, ) self.target_usd = Currency.objects.create( title="USD", shop=self.target_shop, created_by=self.user, ) self.target_afg = Currency.objects.create( title="AFG", shop=self.target_shop, created_by=self.user, ) self.source_purchase = Purchase.objects.create( shop=self.source_shop, currency=self.source_usd, purchase_date=timezone.localdate(), created_by=self.user, ) self.source_purchase_item = PurchaseItem.objects.create( purchase=self.source_purchase, currency=self.source_usd, product=self.product, unit=self.unit, quantity=200, buy_price=Decimal("274.50"), wholesale_price=Decimal("280"), retail_price=Decimal("290"), sale_price=Decimal("290"), created_by=self.user, ) self.source_stock = Stock.objects.create( shop=self.source_shop, purchase_item=self.source_purchase_item, quantity=200, sale_price=Decimal("290"), average_buy_price=Decimal("263.25"), wholesale_price=Decimal("280"), retail_price=Decimal("290"), created_by=self.user, ) # Existing destination AFG stock must NOT absorb a USD transfer. self.target_product = Product.objects.create( name="SPARAKET BOLTS", code="SPB-1", shop=self.target_shop, created_by=self.user, ) target_unit = Unit.objects.create( title="PCS", shop=self.target_shop, created_by=self.user, ) existing_purchase = Purchase.objects.create( shop=self.target_shop, currency=self.target_afg, purchase_date=timezone.localdate(), created_by=self.user, ) existing_item = PurchaseItem.objects.create( purchase=existing_purchase, currency=self.target_afg, product=self.target_product, unit=target_unit, quantity=10, buy_price=Decimal("100"), created_by=self.user, ) self.existing_afg_stock = Stock.objects.create( shop=self.target_shop, purchase_item=existing_item, quantity=10, average_buy_price=Decimal("100"), created_by=self.user, ) self.client.force_authenticate(user=self.user) def _summary(self, shop_id): response = self.client.get( "/api/shop/stocks/summary/", {"shop": shop_id}, HTTP_X_SELECTED_SHOP=str(shop_id), ) self.assertEqual(response.status_code, 200, getattr(response, "data", response.content)) return response.data def test_accept_uses_destination_currency_and_preserves_transfer_value(self): create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.source_usd.id, "transfer_quantity": 140, "buy_price": "274.50", "transfer_date": str(timezone.localdate()), "show_payment_button": False, }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, getattr(create_response, "data", create_response.content)) transfer_id = create_response.data["id"] accept_response = self.client.post( f"/api/shop/stock-transfers/{transfer_id}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, getattr(accept_response, "data", accept_response.content)) transfer = StockTransfer.objects.get(pk=transfer_id) target_stock = transfer.target_stock self.assertIsNotNone(target_stock) target_stock.refresh_from_db() self.existing_afg_stock.refresh_from_db() self.assertEqual(target_stock.shop_id, self.target_shop.id) self.assertEqual(target_stock.quantity, 140) self.assertEqual(target_stock.average_buy_price, Decimal("274.50")) self.assertEqual(target_stock.currency.id, self.target_usd.id) self.assertEqual(target_stock.purchase_item.product.shop_id, self.target_shop.id) # AFG row must remain untouched by the USD transfer. self.assertEqual(self.existing_afg_stock.quantity, 10) self.assertEqual(self.existing_afg_stock.average_buy_price, Decimal("100")) summary = self._summary(self.target_shop.id) values = {row["currency"]: Decimal(row["amount"]) for row in summary["stock_value"]} self.assertEqual(values.get("USD"), Decimal("38430.00")) self.assertEqual(values.get("AFG"), Decimal("1000.00")) def test_second_transfer_merges_into_existing_destination_stock(self): # First transfer creates/uses Takhar stock. first = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.source_usd.id, "transfer_quantity": 10, "buy_price": "200.00", "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(first.status_code, 201, getattr(first, "data", first.content)) accept1 = self.client.post( f"/api/shop/stock-transfers/{first.data['id']}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept1.status_code, 200, getattr(accept1, "data", accept1.content)) first_target_id = StockTransfer.objects.get(pk=first.data["id"]).target_stock_id self.assertIsNotNone(first_target_id) # Second transfer of same product/currency must update the same stock row. second = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.source_usd.id, "transfer_quantity": 5, "buy_price": "300.00", "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(second.status_code, 201, getattr(second, "data", second.content)) accept2 = self.client.post( f"/api/shop/stock-transfers/{second.data['id']}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept2.status_code, 200, getattr(accept2, "data", accept2.content)) second_transfer = StockTransfer.objects.get(pk=second.data["id"]) self.assertEqual(second_transfer.target_stock_id, first_target_id) target = Stock.objects.get(pk=first_target_id) self.assertEqual(target.quantity, 15) # Weighted avg: (10*200 + 5*300) / 15 = 233.3333 (4dp field) self.assertEqual(target.average_buy_price, Decimal("233.3333")) active_same_product = Stock.objects.filter( shop=self.target_shop, is_deleted=False, quantity__gt=0, purchase_item__product__name__iexact=self.product.name, purchase_item__currency=self.target_usd, ) self.assertEqual(active_same_product.count(), 1) customer = Customer.objects.create( name="Takhar Customer", shop=self.target_shop, created_by=self.user, ) sale_response = self.client.post( "/api/shop/sales/", { "shop": self.target_shop.id, "customer": customer.id, "sale_date": str(timezone.localdate()), "currency": self.target_usd.id, "received_amount": "0", "amount": "0", "items": [ { "product": target.purchase_item.product_id, "purchase_item": target.purchase_item_id, "unit": target.purchase_item.unit_id, "quantity": 4, "sale_price": "300.00", } ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual( sale_response.status_code, 201, getattr(sale_response, "data", sale_response.content), ) target.refresh_from_db() self.assertEqual(target.quantity, 11) def test_create_remaps_foreign_currency_id_to_supplier_branch(self): # UI historically listed every shop's currencies; picking Takhar USD # while transferring from Kabul must still reserve Kabul USD stock. create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.target_usd.id, "transfer_quantity": 10, "buy_price": "274.50", "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, getattr(create_response, "data", create_response.content)) transfer = StockTransfer.objects.get(pk=create_response.data["id"]) self.assertEqual(transfer.currency_id, self.source_usd.id) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 190) def test_create_matches_equivalent_afn_afg_style_currency_titles(self): source_dollar_alias = Currency.objects.create( title="US Dollar", shop=self.source_shop, created_by=self.user, ) create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": source_dollar_alias.id, "transfer_quantity": 10, "buy_price": "274.50", "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, create_response.data) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 190) def test_create_resolves_stale_destination_product_id_by_source_product_name(self): create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.target_product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.source_usd.id, "transfer_quantity": 10, "buy_price": "274.50", "transfer_date": str(timezone.localdate()), }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, create_response.data) transfer = StockTransfer.objects.get(pk=create_response.data["id"]) self.assertEqual(transfer.product_id, self.product.id) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 190) class StockTransferDestinationVisibilityTests(TestCase): """Accepted transfers must appear on destination Stock search by product name.""" def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="transfervisible", password="testpass123", is_staff=True, allowed_tabs=["stock", "purchases", "products", "units", "currencies", "category"], ) self.source_shop = Shop.objects.create(name="Kabul Main Branch", owner=self.user) self.target_shop = Shop.objects.create(name="Badakhshan Branch", owner=self.user) self.category = Category.objects.create( name="General", shop=self.source_shop, created_by=self.user, ) # Shared/reused barcode — previously caused wrong destination product match. self.shared_code = "0" self.product = Product.objects.create( name="BROCKEN LINKS", code=self.shared_code, shop=self.source_shop, category=self.category, created_by=self.user, ) self.other_dest_product = Product.objects.create( name="OTHER PART", code=self.shared_code, shop=self.target_shop, created_by=self.user, ) self.unit = Unit.objects.create(title="PCS", shop=self.source_shop, created_by=self.user) self.currency = Currency.objects.create(title="AFG", shop=self.source_shop, created_by=self.user) self.target_currency = Currency.objects.create( title="AFG", shop=self.target_shop, created_by=self.user ) purchase = Purchase.objects.create( shop=self.source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=50, buy_price=Decimal("1050"), wholesale_price=Decimal("1100"), retail_price=Decimal("1200"), sale_price=Decimal("1200"), created_by=self.user, ) self.source_stock = Stock.objects.create( shop=self.source_shop, purchase_item=purchase_item, quantity=50, sale_price=Decimal("1200"), average_buy_price=Decimal("1050"), wholesale_price=Decimal("1100"), retail_price=Decimal("1200"), created_by=self.user, ) other_purchase = Purchase.objects.create( shop=self.target_shop, currency=self.target_currency, purchase_date=timezone.localdate(), created_by=self.user, ) other_item = PurchaseItem.objects.create( purchase=other_purchase, currency=self.target_currency, product=self.other_dest_product, unit=Unit.objects.create(title="PCS", shop=self.target_shop, created_by=self.user), quantity=3, buy_price=Decimal("10"), created_by=self.user, ) self.other_stock = Stock.objects.create( shop=self.target_shop, purchase_item=other_item, quantity=3, average_buy_price=Decimal("10"), created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_accept_does_not_attach_to_different_product_with_same_code(self): create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.currency.id, "transfer_quantity": 10, "buy_price": "1050.00", "transfer_date": str(timezone.localdate()), "show_payment_button": False, }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, getattr(create_response, "data", create_response.content)) accept_response = self.client.post( f"/api/shop/stock-transfers/{create_response.data['id']}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, getattr(accept_response, "data", accept_response.content)) transfer = StockTransfer.objects.get(pk=create_response.data["id"]) self.assertIsNotNone(transfer.target_stock) dest_product = transfer.target_stock.purchase_item.product self.assertEqual(dest_product.name, "BROCKEN LINKS") self.assertEqual(dest_product.shop_id, self.target_shop.id) self.assertNotEqual(dest_product.id, self.other_dest_product.id) self.other_stock.refresh_from_db() self.assertEqual(self.other_stock.quantity, 3) stock_list = self.client.get( "/api/shop/stocks/", {"shop": self.target_shop.id, "search": "BROCKEN LINKS"}, HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(stock_list.status_code, 200, getattr(stock_list, "data", stock_list.content)) results = stock_list.data.get("results", stock_list.data) self.assertTrue(any(int(row.get("quantity") or 0) == 10 for row in results)) def test_accept_heals_stock_with_missing_purchase_item(self): orphan = Stock.objects.create( shop=self.target_shop, purchase_item=None, quantity=0, average_buy_price=Decimal("1050"), created_by=self.user, ) # Seed a destination product so find-target can resolve by name after create. Product.objects.create( name="BROCKEN LINKS", code=self.shared_code, shop=self.target_shop, created_by=self.user, ) # Orphan row has no purchase_item so currency/product filters won't find it; # accept should still create a listable destination stock. create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.currency.id, "transfer_quantity": 7, "buy_price": "1050.00", "transfer_date": str(timezone.localdate()), "show_payment_button": False, }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, getattr(create_response, "data", create_response.content)) accept_response = self.client.post( f"/api/shop/stock-transfers/{create_response.data['id']}/accept/", {}, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, getattr(accept_response, "data", accept_response.content)) transfer = StockTransfer.objects.get(pk=create_response.data["id"]) self.assertIsNotNone(transfer.target_stock) self.assertIsNotNone(transfer.target_stock.purchase_item_id) self.assertFalse(transfer.target_stock.purchase_item.is_deleted) self.assertEqual(transfer.target_stock.purchase_item.product.name, "BROCKEN LINKS") self.assertEqual(transfer.target_stock.quantity, 7) stock_list = self.client.get( "/api/shop/stocks/", {"shop": self.target_shop.id, "search": "BROCKEN LINKS"}, HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) results = stock_list.data.get("results", stock_list.data) self.assertTrue(any(int(row.get("quantity") or 0) >= 7 for row in results)) orphan.refresh_from_db() # Orphan without PI should remain untouched (not used as merge target). self.assertEqual(orphan.quantity, 0) class StockTransferPaymentSupplierTests(TestCase): """Payment checkbox must control Transfer-from supplier on destination stock.""" def setUp(self): self.client = APIClient() self.user = User.objects.create_user( username="transferpayment", password="testpass123", is_staff=True, allowed_tabs=[ "stock", "stock_transfer", "sales", "purchases", "products", "units", "currencies", "category", "suppliers", ], ) self.source_shop = Shop.objects.create(name="Kabul Main Branch", owner=self.user) self.target_shop = Shop.objects.create(name="Takhar Branch", owner=self.user) self.category = Category.objects.create( name="General", shop=self.source_shop, created_by=self.user, ) self.product = Product.objects.create( name="DX AIR FILTER", code="320", shop=self.source_shop, category=self.category, created_by=self.user, ) self.unit = Unit.objects.create(title="PCS", shop=self.source_shop, created_by=self.user) self.currency = Currency.objects.create(title="AFG", shop=self.source_shop, created_by=self.user) self.target_currency = Currency.objects.create( title="AFG", shop=self.target_shop, created_by=self.user ) purchase = Purchase.objects.create( shop=self.source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=200, buy_price=Decimal("950"), wholesale_price=Decimal("2000"), retail_price=Decimal("2200"), sale_price=Decimal("950"), created_by=self.user, ) self.source_stock = Stock.objects.create( shop=self.source_shop, purchase_item=purchase_item, quantity=200, sale_price=Decimal("950"), average_buy_price=Decimal("950"), wholesale_price=Decimal("2000"), retail_price=Decimal("2200"), created_by=self.user, ) self.client.force_authenticate(user=self.user) def _create_and_accept(self, quantity, show_payment, accept_payment=None): create_response = self.client.post( "/api/shop/stock-transfers/", { "product": self.product.id, "supplier_branch": self.source_shop.id, "customer_branch": self.target_shop.id, "currency": self.currency.id, "transfer_quantity": quantity, "buy_price": "950.00", "sale_price": "950.00", "transfer_date": str(timezone.localdate()), "show_payment_button": show_payment, }, format="json", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(create_response.status_code, 201, getattr(create_response, "data", create_response.content)) payload = {} if accept_payment is not None: payload["show_payment_button"] = accept_payment accept_response = self.client.post( f"/api/shop/stock-transfers/{create_response.data['id']}/accept/", payload, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(accept_response.status_code, 200, getattr(accept_response, "data", accept_response.content)) return StockTransfer.objects.get(pk=create_response.data["id"]) def test_payment_off_has_no_transfer_supplier_on_stock(self): transfer = self._create_and_accept(50, show_payment=False) target = transfer.target_stock self.assertIsNotNone(target) self.assertIsNotNone(target.purchase_item_id) supplier = target.purchase_item.purchase.supplier if target.purchase_item.purchase_id else None self.assertIsNone(supplier) self.assertFalse( Supplier.objects.filter( shop=self.target_shop, name__iexact="Transfer from Kabul Main Branch", ).exists() ) def test_payment_on_shows_transfer_supplier_on_stock(self): transfer = self._create_and_accept(50, show_payment=True) target = transfer.target_stock self.assertIsNotNone(target) supplier = target.purchase_item.purchase.supplier self.assertIsNotNone(supplier) self.assertEqual(supplier.name, "Transfer from Kabul Main Branch") def test_transfer_generated_sale_cannot_restore_stock_when_deleted_from_sales(self): transfer = self._create_and_accept(50, show_payment=True) self.source_stock.refresh_from_db() source_quantity = self.source_stock.quantity transfer_sale = transfer.supplier_branch.sales.get( notes=f"Stock transfer sale for {transfer.transfer_id}" ) delete_response = self.client.delete( f"/api/shop/sales/{transfer_sale.id}/", HTTP_X_SELECTED_SHOP=str(self.source_shop.id), ) self.assertEqual(delete_response.status_code, 400, delete_response.data) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, source_quantity) def test_payment_off_purchase_qty_is_transfer_qty_not_full_stock(self): dest_product = Product.objects.create( name="DX AIR FILTER", code="149", shop=self.target_shop, created_by=self.user, ) dest_unit = Unit.objects.create(title="PCS", shop=self.target_shop, created_by=self.user) transfer_supplier = Supplier.objects.create( shop=self.target_shop, name="Transfer from Kabul Main Branch", created_by=self.user, ) dest_purchase = Purchase.objects.create( shop=self.target_shop, supplier=transfer_supplier, currency=self.target_currency, purchase_date=timezone.localdate(), created_by=self.user, ) dest_item = PurchaseItem.objects.create( purchase=dest_purchase, currency=self.target_currency, product=dest_product, unit=dest_unit, quantity=57, buy_price=Decimal("950"), created_by=self.user, ) Stock.objects.create( shop=self.target_shop, purchase_item=dest_item, quantity=57, average_buy_price=Decimal("950"), sale_price=Decimal("950"), created_by=self.user, ) transfer = self._create_and_accept(50, show_payment=False) target = transfer.target_stock target.refresh_from_db() self.assertEqual(target.quantity, 107) self.assertEqual(int(target.purchase_item.quantity), 50) self.assertIsNone(target.purchase_item.purchase.supplier) def test_accept_payload_can_override_payment_flag_off(self): # Created with Payment ON, accepted with Payment OFF → no supplier. transfer = self._create_and_accept(40, show_payment=True, accept_payment=False) self.assertFalse(transfer.show_payment_button) supplier = transfer.target_stock.purchase_item.purchase.supplier self.assertIsNone(supplier) def test_merges_duplicate_same_name_different_code_stocks(self): # Existing Takhar stock for same product name but different code (duplicate Product). dest_product_a = Product.objects.create( name="DX AIR FILTER", code="149", shop=self.target_shop, created_by=self.user, ) dest_product_b = Product.objects.create( name="DX AIR FILTER", code="320", shop=self.target_shop, created_by=self.user, ) dest_unit = Unit.objects.create(title="PCS", shop=self.target_shop, created_by=self.user) existing_purchase = Purchase.objects.create( shop=self.target_shop, supplier=None, currency=self.target_currency, amount=0, purchase_date=timezone.localdate(), created_by=self.user, ) existing_item = PurchaseItem.objects.create( purchase=existing_purchase, currency=self.target_currency, product=dest_product_a, unit=dest_unit, quantity=55, buy_price=Decimal("950"), created_by=self.user, ) existing_stock = Stock.objects.create( shop=self.target_shop, purchase_item=existing_item, quantity=55, average_buy_price=Decimal("950"), created_by=self.user, ) # Stale second row that previously got a Transfer supplier label. transfer_supplier = Supplier.objects.create( shop=self.target_shop, name="Transfer from Kabul Main Branch", created_by=self.user, ) stale_purchase = Purchase.objects.create( shop=self.target_shop, supplier=transfer_supplier, currency=self.target_currency, amount=Decimal("47500"), purchase_date=timezone.localdate(), created_by=self.user, ) stale_item = PurchaseItem.objects.create( purchase=stale_purchase, currency=self.target_currency, product=dest_product_b, unit=dest_unit, quantity=50, buy_price=Decimal("950"), created_by=self.user, ) stale_stock = Stock.objects.create( shop=self.target_shop, purchase_item=stale_item, quantity=50, average_buy_price=Decimal("950"), created_by=self.user, ) transfer = self._create_and_accept(10, show_payment=False) target = transfer.target_stock self.assertIsNotNone(target) active = Stock.objects.filter( shop=self.target_shop, is_deleted=False, quantity__gt=0, purchase_item__product__name__iexact="DX AIR FILTER", ) self.assertEqual(active.count(), 1) surviving = active.first() self.assertEqual(surviving.quantity, 55 + 50 + 10) self.assertIsNone(surviving.purchase_item.purchase.supplier) existing_stock.refresh_from_db() stale_stock.refresh_from_db() # At least one of the pre-existing rows was merged away. self.assertTrue(existing_stock.is_deleted or stale_stock.is_deleted or surviving.id in (existing_stock.id, stale_stock.id)) def test_payment_on_changes_source_and_destination_qty_once(self): transfer = self._create_and_accept(50, show_payment=True) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 150) self.assertEqual(transfer.target_stock.quantity, 50) dest_total = sum( int(row.quantity or 0) for row in Stock.objects.filter( shop=self.target_shop, is_deleted=False, purchase_item__product__name__iexact="DX AIR FILTER", ) ) self.assertEqual(dest_total, 50) second = self._create_and_accept(10, show_payment=True) self.source_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 140) dest_total = sum( int(row.quantity or 0) for row in Stock.objects.filter( shop=self.target_shop, is_deleted=False, purchase_item__product__name__iexact="DX AIR FILTER", ) ) self.assertEqual(dest_total, 60) self.assertEqual(second.target_stock.quantity, 60) def test_transfer_purchase_cannot_change_stock_from_purchases_api(self): transfer = self._create_and_accept(50, show_payment=True) purchase = transfer.target_stock.purchase_item.purchase purchase_item = transfer.target_stock.purchase_item dest_qty = transfer.target_stock.quantity response = self.client.put( f"/api/shop/purchases/{purchase.id}/", { "shop": self.target_shop.id, "supplier": purchase.supplier_id, "currency": self.target_currency.id, "amount": "0", "received_amount": "0", "purchase_date": str(timezone.localdate()), "items": [ { "id": purchase_item.id, "product": purchase_item.product_id, "unit": purchase_item.unit_id, "quantity": 80, "buy_price": str(purchase_item.buy_price), } ], }, format="json", HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(response.status_code, 400, response.data) transfer.target_stock.refresh_from_db() self.assertEqual(transfer.target_stock.quantity, dest_qty) def test_payment_off_transfer_purchase_hidden_from_purchases_list(self): off_transfer = self._create_and_accept(50, show_payment=False) off_purchase_id = off_transfer.target_stock.purchase_item.purchase_id on_transfer = self._create_and_accept(10, show_payment=True) on_purchase_id = on_transfer.target_stock.purchase_item.purchase_id response = self.client.get( "/api/shop/purchases/", {"shop": self.target_shop.id, "page_size": 100}, HTTP_X_SELECTED_SHOP=str(self.target_shop.id), ) self.assertEqual(response.status_code, 200, response.data) ids = [ row["id"] for row in (response.data.get("results") or response.data) ] self.assertNotIn(off_purchase_id, ids) self.assertIn(on_purchase_id, ids) class HealStockQuantitiesCommandTests(TestCase): def setUp(self): self.user = User.objects.create_user( username="healqty", password="testpass123", is_staff=True, is_superuser=True, ) self.source_shop = Shop.objects.create(name="Kabul Main Branch", owner=self.user) self.target_shop = Shop.objects.create(name="Takhar Branch", owner=self.user) self.product = Product.objects.create( name="MASTER PIN", code="MP-1", shop=self.source_shop, created_by=self.user, ) dest_product = Product.objects.create( name="MASTER PIN", code="MP-1", shop=self.target_shop, created_by=self.user, ) self.unit = Unit.objects.create(title="PCS", shop=self.source_shop, created_by=self.user) dest_unit = Unit.objects.create(title="PCS", shop=self.target_shop, created_by=self.user) self.currency = Currency.objects.create(title="AFG", shop=self.source_shop, created_by=self.user) dest_currency = Currency.objects.create(title="AFG", shop=self.target_shop, created_by=self.user) purchase = Purchase.objects.create( shop=self.source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=self.product, unit=self.unit, quantity=20, buy_price=Decimal("10"), created_by=self.user, ) self.source_stock = Stock.objects.create( shop=self.source_shop, purchase_item=purchase_item, quantity=12, average_buy_price=Decimal("10"), created_by=self.user, ) transfer_supplier = Supplier.objects.create( shop=self.target_shop, name="Transfer from Kabul Main Branch", created_by=self.user, ) transfer_purchase = Purchase.objects.create( shop=self.target_shop, supplier=transfer_supplier, currency=dest_currency, purchase_date=timezone.localdate(), created_by=self.user, ) transfer_item = PurchaseItem.objects.create( purchase=transfer_purchase, currency=dest_currency, product=dest_product, unit=dest_unit, quantity=8, buy_price=Decimal("10"), created_by=self.user, ) self.target_stock = Stock.objects.create( shop=self.target_shop, purchase_item=transfer_item, quantity=16, average_buy_price=Decimal("10"), created_by=self.user, ) StockTransfer.objects.create( transfer_id="TRF-HEAL-001", product=self.product, supplier_branch=self.source_shop, customer_branch=self.target_shop, source_stock=self.source_stock, target_stock=self.target_stock, currency=self.currency, transfer_quantity=8, old_quantity=20, new_quantity=12, transfer_date=timezone.localdate(), status=StockTransfer.Status.ACCEPTED, created_by=self.user, ) def test_apply_does_not_count_transfer_purchase_twice(self): from django.core.management import call_command dest_supplier = Supplier.objects.create( shop=self.target_shop, name="Local Supplier", created_by=self.user, ) dest_purchase = Purchase.objects.create( shop=self.target_shop, supplier=dest_supplier, currency=Currency.objects.get(shop=self.target_shop, title="AFG"), purchase_date=timezone.localdate(), created_by=self.user, ) dest_item = PurchaseItem.objects.create( purchase=dest_purchase, product=Product.objects.get(shop=self.target_shop, name="MASTER PIN"), unit=Unit.objects.get(shop=self.target_shop, title="PCS"), quantity=10, buy_price=Decimal("10"), created_by=self.user, ) self.target_stock.purchase_item = dest_item self.target_stock.quantity = 26 self.target_stock.save(update_fields=["purchase_item", "quantity", "updated_at"]) call_command("heal_stock_quantities", apply=True) self.source_stock.refresh_from_db() self.target_stock.refresh_from_db() self.assertEqual(self.source_stock.quantity, 12) self.assertEqual(self.target_stock.quantity, 26) def test_apply_raises_missing_qty_only(self): from django.core.management import call_command dest_supplier = Supplier.objects.create( shop=self.target_shop, name="Local Supplier", created_by=self.user, ) dest_purchase = Purchase.objects.create( shop=self.target_shop, supplier=dest_supplier, currency=Currency.objects.get(shop=self.target_shop, title="AFG"), purchase_date=timezone.localdate(), created_by=self.user, ) dest_item = PurchaseItem.objects.create( purchase=dest_purchase, product=Product.objects.get(shop=self.target_shop, name="MASTER PIN"), unit=Unit.objects.get(shop=self.target_shop, title="PCS"), quantity=10, buy_price=Decimal("10"), created_by=self.user, ) self.target_stock.purchase_item = dest_item self.target_stock.quantity = 10 self.target_stock.save(update_fields=["purchase_item", "quantity", "updated_at"]) call_command("heal_stock_quantities", apply=True) self.target_stock.refresh_from_db() self.assertEqual(self.target_stock.quantity, 18) def _make_sold_more_stock(self): product = Product.objects.create( name="SWING GIRA", code="SG-1", shop=self.source_shop, created_by=self.user, ) purchase = Purchase.objects.create( shop=self.source_shop, currency=self.currency, purchase_date=timezone.localdate(), created_by=self.user, ) purchase_item = PurchaseItem.objects.create( purchase=purchase, currency=self.currency, product=product, unit=self.unit, quantity=1, buy_price=Decimal("10"), retail_price=Decimal("20"), sale_price=Decimal("20"), created_by=self.user, ) stock = Stock.objects.create( shop=self.source_shop, purchase_item=purchase_item, quantity=5, average_buy_price=Decimal("10"), created_by=self.user, ) customer = Customer.objects.create( name="Walk-in", shop=self.source_shop, created_by=self.user, ) sale = Sale.objects.create( shop=self.source_shop, customer=customer, currency=self.currency, sale_date=timezone.localdate(), created_by=self.user, ) SaleItem.objects.create( sale=sale, product=product, purchase_item=purchase_item, quantity=10, unit=self.unit, buy_price=Decimal("10"), sale_price=Decimal("20"), created_by=self.user, ) return stock, product def test_add_opening_creates_bill_without_changing_stock(self): from django.core.management import call_command stock, product = self._make_sold_more_stock() call_command( "heal_stock_quantities", add_opening=True, apply=True, product_name="SWING GIRA", ) stock.refresh_from_db() self.assertEqual(stock.quantity, 5) opening = PurchaseItem.objects.filter( purchase__shop=self.source_shop, purchase__supplier__name="Opening Stock", product=product, is_deleted=False, ) self.assertEqual(opening.count(), 1) self.assertEqual(opening.first().quantity, 14) self.assertEqual(opening.first().buy_price, Decimal("0")) call_command( "heal_stock_quantities", add_opening=True, apply=True, product_name="SWING GIRA", ) self.assertEqual( PurchaseItem.objects.filter( purchase__shop=self.source_shop, purchase__supplier__name="Opening Stock", product=product, is_deleted=False, ).count(), 1, ) stock.refresh_from_db() self.assertEqual(stock.quantity, 5) def test_add_opening_dry_run_does_not_create_bill(self): from django.core.management import call_command stock, product = self._make_sold_more_stock() call_command( "heal_stock_quantities", add_opening=True, product_name="SWING GIRA", ) stock.refresh_from_db() self.assertEqual(stock.quantity, 5) self.assertFalse( PurchaseItem.objects.filter( purchase__supplier__name="Opening Stock", product=product, ).exists() ) class DualCurrencyOpeningAndCrossCurrencySettlementTests(TestCase): """ Tests for dual-currency opening balances (AFN & USD), preserving original opening balances, customer & supplier cross-currency debt settlement, overpayments/credit, and historical exchange rate immutability. """ def setUp(self): from shop_management.models.data.utils import ExchangeRate from shop_management.models.data.supplier_balance_payment import SupplierBalancePayment self.ExchangeRate = ExchangeRate self.SupplierBalancePayment = SupplierBalancePayment self.client = APIClient() self.user = User.objects.create_user( username="dualcurrencyadmin", password="testpass123", is_staff=True, is_superuser=True, allowed_tabs=[ "customers", "suppliers", "sales", "purchases", "accounts", "currencies", "reports", "dashboard", ], ) self.shop = Shop.objects.create(name="Kabul Main", owner=self.user) self.afn = Currency.objects.create( title="AFN", shop=self.shop, created_by=self.user ) self.usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user ) self.afn_account = Account.objects.create( shop=self.shop, name="AFN Cash", currency=self.afn, opening_balance=Decimal("1000000.00"), allow_overdraft=True, ) self.usd_account = Account.objects.create( shop=self.shop, name="USD Cash", currency=self.usd, opening_balance=Decimal("50000.00"), allow_overdraft=True, ) # 1 USD = 70 AFN self.rate_row = ExchangeRate.objects.create( shop=self.shop, from_currency=self.usd, to_currency=self.afn, rate=Decimal("70.000000"), rate_date=timezone.localdate(), ) self.client.force_authenticate(user=self.user) def test_01_customer_with_only_afn_opening_balance(self): customer = Customer.objects.create( name="Customer AFN Only", shop=self.shop, opening_balance_afn=Decimal("50000.00"), opening_balance_usd=Decimal("0.00"), created_by=self.user, ) self.assertEqual(customer.get_total_owed_by_currency(), {"AFN": 50000.0}) self.assertEqual(customer.get_original_opening_balances_by_currency(), {"AFN": 50000.0}) self.assertEqual(customer.remaining_opening_balance_afn, Decimal("50000.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("0.00")) def test_02_customer_with_only_usd_opening_balance(self): customer = Customer.objects.create( name="Customer USD Only", shop=self.shop, opening_balance_afn=Decimal("0.00"), opening_balance_usd=Decimal("250.00"), created_by=self.user, ) self.assertEqual(customer.get_total_owed_by_currency(), {"USD": 250.0}) self.assertEqual(customer.get_original_opening_balances_by_currency(), {"USD": 250.0}) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("250.00")) def test_03_customer_with_both_afn_and_usd_opening_balances(self): customer = Customer.objects.create( name="Customer Dual Opening", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) self.assertEqual( customer.get_total_owed_by_currency(), {"AFN": 10000.0, "USD": 100.0}, ) self.assertEqual( customer.get_original_opening_balances_by_currency(), {"AFN": 10000.0, "USD": 100.0}, ) def test_04_supplier_with_only_afn_opening_balance(self): supplier = Supplier.objects.create( name="Supplier AFN Only", shop=self.shop, opening_balance_afn=Decimal("40000.00"), opening_balance_usd=Decimal("0.00"), created_by=self.user, ) self.assertEqual(supplier.get_total_owed_by_currency(), {"AFN": 40000.0}) self.assertEqual(supplier.get_original_opening_balances_by_currency(), {"AFN": 40000.0}) def test_05_supplier_with_only_usd_opening_balance(self): supplier = Supplier.objects.create( name="Supplier USD Only", shop=self.shop, opening_balance_afn=Decimal("0.00"), opening_balance_usd=Decimal("300.00"), created_by=self.user, ) self.assertEqual(supplier.get_total_owed_by_currency(), {"USD": 300.0}) self.assertEqual(supplier.get_original_opening_balances_by_currency(), {"USD": 300.0}) def test_06_supplier_with_both_afn_and_usd_opening_balances(self): supplier = Supplier.objects.create( name="Supplier Dual Opening", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) self.assertEqual( supplier.get_total_owed_by_currency(), {"AFN": 10000.0, "USD": 100.0}, ) def test_07_partial_payment_preserves_original_opening_balance(self): # Customer: Original opening balance 50,000 AFN, pay 20,000 AFN -> 30,000 AFN remaining customer = Customer.objects.create( name="Preserve Opening Customer", shop=self.shop, opening_balance_afn=Decimal("50000.00"), opening_balance_usd=Decimal("200.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "20000.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) customer.refresh_from_db() # Original opening balance MUST remain 50,000 AFN! self.assertEqual(customer.opening_balance_afn, Decimal("50000.00")) self.assertEqual(customer.opening_balance_afn_paid, Decimal("20000.00")) self.assertEqual(customer.remaining_opening_balance_afn, Decimal("30000.00")) self.assertEqual( customer.get_total_owed_by_currency(), {"AFN": 30000.0, "USD": 200.0}, ) # Supplier: Original opening balance 50,000 AFN, pay 20,000 AFN -> 30,000 AFN remaining supplier = Supplier.objects.create( name="Preserve Opening Supplier", shop=self.shop, opening_balance_afn=Decimal("50000.00"), opening_balance_usd=Decimal("150.00"), created_by=self.user, ) s_resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "20000.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(s_resp.status_code, 201, s_resp.data) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn, Decimal("50000.00")) self.assertEqual(supplier.opening_balance_afn_paid, Decimal("20000.00")) self.assertEqual(supplier.remaining_opening_balance_afn, Decimal("30000.00")) self.assertEqual( supplier.get_total_owed_by_currency(), {"AFN": 30000.0, "USD": 150.0}, ) def test_08_full_payment_in_same_currency(self): supplier = Supplier.objects.create( name="Full Same Currency Supplier", shop=self.shop, opening_balance_afn=Decimal("15000.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "15000.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn, Decimal("15000.00")) self.assertEqual(supplier.opening_balance_afn_paid, Decimal("15000.00")) self.assertEqual(supplier.remaining_opening_balance_afn, Decimal("0.00")) self.assertEqual(supplier.get_total_owed_by_currency(), {}) def test_09_payment_in_afn_settles_afn_and_partially_settles_usd(self): # Supplier owed 10,000 AFN and 100 USD; pay 13,500 AFN at 70 AFN/USD -> # settles 10,000 AFN and 50 USD (using 3,500 AFN), leaving 50 USD! supplier = Supplier.objects.create( name="Partial Cascade Supplier", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "13500.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn, Decimal("10000.00")) self.assertEqual(supplier.opening_balance_afn_paid, Decimal("10000.00")) self.assertEqual(supplier.opening_balance_usd, Decimal("100.00")) self.assertEqual(supplier.opening_balance_usd_paid, Decimal("50.00")) self.assertEqual(supplier.get_total_owed_by_currency(), {"USD": 50.0}) def test_10_payment_in_afn_settles_both_afn_and_usd_completely(self): # Exact user scenario: Supplier owed 10,000 AFN and 100 USD, paid 17,000 AFN at 70 AFN/USD supplier = Supplier.objects.create( name="Complete Cascade Supplier", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "17000.00", "settlement_exchange_rate": "1", "secondary_exchange_rate": "70", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn, Decimal("10000.00")) self.assertEqual(supplier.opening_balance_afn_paid, Decimal("10000.00")) self.assertEqual(supplier.opening_balance_usd, Decimal("100.00")) self.assertEqual(supplier.opening_balance_usd_paid, Decimal("100.00")) self.assertEqual(supplier.get_total_owed_by_currency(), {}) payment = self.SupplierBalancePayment.objects.get(pk=resp.data["id"]) self.assertEqual(payment.allocated_to_afn, Decimal("10000.00")) self.assertEqual(payment.allocated_to_usd, Decimal("100.00")) self.assertEqual(payment.allocated_to_opening_afn, Decimal("10000.00")) self.assertEqual(payment.allocated_to_opening_usd, Decimal("100.00")) self.assertEqual(payment.unallocated_amount, Decimal("0.00")) def test_11_payment_in_usd_settles_usd_and_afn_debt(self): # Customer owed 7,000 AFN and 100 USD; pays 180 USD -> settles 100 USD and 5,600 AFN (80 USD * 70) customer = Customer.objects.create( name="USD Cascade Customer", shop=self.shop, opening_balance_afn=Decimal("7000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.usd_account.id, "currency": self.usd.id, "settlement_currency": self.usd.id, "amount": "180.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) customer.refresh_from_db() self.assertEqual(customer.opening_balance_usd_paid, Decimal("100.00")) self.assertEqual(customer.opening_balance_afn_paid, Decimal("5600.00")) self.assertEqual(customer.get_total_owed_by_currency(), {"AFN": 1400.0}) def test_12_overpayment_creates_unallocated_credit_for_customer_and_supplier(self): # Supplier owed 10,000 AFN and 100 USD; pays 20,000 AFN at 70 AFN/USD -> # 10,000 AFN + 7,000 AFN (for 100 USD) = 17,000 AFN used; 3,000 AFN unallocated credit! supplier = Supplier.objects.create( name="Overpaid Supplier", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "20000.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) supplier.refresh_from_db() self.assertEqual(supplier.get_total_owed_by_currency(), {}) self.assertEqual(supplier.get_credit_by_currency(), {"AFN": 3000.0}) def test_13_historical_payment_records_unaffected_by_later_exchange_rate_changes(self): customer = Customer.objects.create( name="Historical Rate Customer", shop=self.shop, opening_balance_afn=Decimal("10000.00"), opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "17000.00", "settlement_exchange_rate": "1", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", HTTP_X_SELECTED_SHOP=str(self.shop.id), ) self.assertEqual(resp.status_code, 201, resp.data) payment = CustomerBalancePayment.objects.get(pk=resp.data["id"]) saved_secondary_rate = payment.secondary_exchange_rate # Change exchange rate in DB from 70 to 90 self.rate_row.rate = Decimal("90.000000") self.rate_row.save(update_fields=["rate"]) customer.refresh_from_db() payment.refresh_from_db() self.assertEqual(customer.get_total_owed_by_currency(), {}) self.assertEqual(payment.allocated_to_afn, Decimal("10000.00")) self.assertEqual(payment.allocated_to_usd, Decimal("100.00")) self.assertEqual(payment.secondary_exchange_rate, saved_secondary_rate) class DualCurrencyFinancialIntegrityTests(TestCase): """ Permanent regression tests for dual-currency financial integrity: Safe cancellation/voiding, exact allocation reversal, immutability of posted payments, idempotency/duplicate submission protection, and accounting transaction consistency. """ def setUp(self): from shop_management.models.data.customer import Customer from shop_management.models.data.supplier import Supplier from shop_management.models.data.shop import Shop from shop_management.models.data.shop_account import Account from shop_management.models.data.utils import Currency, ExchangeRate, classify_currency_code from shop_management.models.data.customer_balance_payment import ( CustomerBalancePayment, CustomerBalancePaymentAllocation, ) from shop_management.models.data.supplier_balance_payment import ( SupplierBalancePayment, SupplierBalancePaymentAllocation, ) from shop_management.models.data.sale import Sale, SaleItem from shop_management.models.data.purchase import Purchase, PurchaseItem self.SaleItem = SaleItem self.PurchaseItem = PurchaseItem from shop_management.models.data.customer_debtor import CustomerDebtor from shop_management.models.data.account_transaction import ( AccountTransaction, AccountTransactionType, ) self.Customer = Customer self.Supplier = Supplier self.CustomerBalancePayment = CustomerBalancePayment self.CustomerBalancePaymentAllocation = CustomerBalancePaymentAllocation self.SupplierBalancePayment = SupplierBalancePayment self.SupplierBalancePaymentAllocation = SupplierBalancePaymentAllocation self.Sale = Sale self.Purchase = Purchase self.CustomerDebtor = CustomerDebtor self.AccountTransaction = AccountTransaction self.AccountTransactionType = AccountTransactionType self.classify_currency_code = classify_currency_code self.client = APIClient() self.user = User.objects.create_user( username="integrityadmin", password="testpass123", is_staff=True, is_superuser=True, allowed_tabs=[ "customers", "suppliers", "sales", "sales_payments", "purchases", "purchase_payments", "accounts", "currencies", "reports", "dashboard", ], ) self.shop = Shop.objects.create(name="Integrity Main Shop", owner=self.user) self.afn = Currency.objects.create( title="AFN", shop=self.shop, created_by=self.user ) self.usd = Currency.objects.create( title="USD", shop=self.shop, created_by=self.user ) self.afn_account = Account.objects.create( shop=self.shop, name="AFN Vault", currency=self.afn, opening_balance=Decimal("1000000.00"), allow_overdraft=True, ) self.usd_account = Account.objects.create( shop=self.shop, name="USD Vault", currency=self.usd, opening_balance=Decimal("50000.00"), allow_overdraft=True, ) # 1 USD = 70 AFN self.rate_usd_to_afn = ExchangeRate.objects.create( shop=self.shop, from_currency=self.usd, to_currency=self.afn, rate=Decimal("70.000000"), rate_date=timezone.localdate(), ) self.client.force_authenticate(user=self.user) # 1. AFN opening balance payment and cancellation def test_01_afn_opening_balance_payment_and_cancellation(self): # Customer customer = self.Customer.objects.create( name="Cust OB AFN", shop=self.shop, opening_balance_afn=Decimal("10000.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "4000.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) customer.refresh_from_db() self.assertEqual(customer.opening_balance_afn_paid, Decimal("4000.00")) self.assertEqual(customer.remaining_opening_balance_afn, Decimal("6000.00")) self.assertEqual(payment.allocations.count(), 1) self.assertEqual(payment.allocations.first().allocated_amount, Decimal("4000.00")) # Void customer payment del_resp = self.client.delete(f"/api/shop/customer-balance-payments/{payment.id}/") self.assertEqual(del_resp.status_code, 204) payment.refresh_from_db() customer.refresh_from_db() self.assertTrue(payment.is_void) self.assertEqual(customer.opening_balance_afn_paid, Decimal("0.00")) self.assertEqual(customer.remaining_opening_balance_afn, Decimal("10000.00")) self.assertTrue(payment.allocations.first().is_reversed) # Supplier supplier = self.Supplier.objects.create( name="Supp OB AFN", shop=self.shop, opening_balance_afn=Decimal("15000.00"), created_by=self.user, ) s_resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "6000.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(s_resp.status_code, 201) s_payment = self.SupplierBalancePayment.objects.get(pk=s_resp.data["id"]) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn_paid, Decimal("6000.00")) s_del = self.client.delete(f"/api/shop/supplier-balance-payments/{s_payment.id}/") self.assertEqual(s_del.status_code, 204) s_payment.refresh_from_db() supplier.refresh_from_db() self.assertTrue(s_payment.is_void) self.assertEqual(supplier.opening_balance_afn_paid, Decimal("0.00")) self.assertEqual(supplier.remaining_opening_balance_afn, Decimal("15000.00")) # 2. USD opening balance payment and cancellation def test_02_usd_opening_balance_payment_and_cancellation(self): customer = self.Customer.objects.create( name="Cust OB USD", shop=self.shop, opening_balance_usd=Decimal("500.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.usd_account.id, "currency": self.usd.id, "settlement_currency": self.usd.id, "amount": "200.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) customer.refresh_from_db() self.assertEqual(customer.opening_balance_usd_paid, Decimal("200.00")) payment.void(user=self.user, reason="Test void USD") payment.refresh_from_db() customer.refresh_from_db() self.assertTrue(payment.is_void) self.assertEqual(customer.opening_balance_usd_paid, Decimal("0.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("500.00")) # 3. AFN payment settling USD debt def test_03_afn_payment_settling_usd_debt(self): customer = self.Customer.objects.create( name="Cust USD Debt", shop=self.shop, opening_balance_usd=Decimal("100.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.usd.id, "settlement_exchange_rate": "70", "amount": "7000.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) customer.refresh_from_db() self.assertEqual(customer.opening_balance_usd_paid, Decimal("100.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("0.00")) # Void restores the USD debt payment.void(user=self.user) customer.refresh_from_db() self.assertEqual(customer.opening_balance_usd_paid, Decimal("0.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("100.00")) # 4. USD payment settling AFN debt def test_04_usd_payment_settling_afn_debt(self): supplier = self.Supplier.objects.create( name="Supp AFN Debt", shop=self.shop, opening_balance_afn=Decimal("7000.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.usd_account.id, "currency": self.usd.id, "settlement_currency": self.afn.id, "settlement_exchange_rate": "70", "amount": "100.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.SupplierBalancePayment.objects.get(pk=resp.data["id"]) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn_paid, Decimal("7000.00")) payment.void(user=self.user) supplier.refresh_from_db() self.assertEqual(supplier.opening_balance_afn_paid, Decimal("0.00")) self.assertEqual(supplier.remaining_opening_balance_afn, Decimal("7000.00")) # 5. Payment settling multiple debts and currencies def test_05_payment_settling_multiple_debts_and_currencies(self): customer = self.Customer.objects.create( name="Cust Multi Debt", shop=self.shop, opening_balance_afn=Decimal("5000.00"), opening_balance_usd=Decimal("50.00"), created_by=self.user, ) # Pay 8500 AFN: 5000 settles AFN opening, 3500 AFN @ 70 settles 50 USD opening resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "8500.00", "secondary_exchange_rate": "70", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) customer.refresh_from_db() self.assertEqual(customer.remaining_opening_balance_afn, Decimal("0.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("0.00")) self.assertEqual(payment.allocations.count(), 2) # Void restores both AFN and USD debts payment.void(user=self.user) customer.refresh_from_db() self.assertEqual(customer.remaining_opening_balance_afn, Decimal("5000.00")) self.assertEqual(customer.remaining_opening_balance_usd, Decimal("500.00") if False else Decimal("50.00")) # 6. Payment allocated to sale/purchase records def test_06_payment_allocated_to_sale_and_purchase_records(self): customer = self.Customer.objects.create(name="Cust Sale", shop=self.shop, created_by=self.user) sale = self.Sale.objects.create( customer=customer, shop=self.shop, currency=self.afn, amount=Decimal("0.00"), received_amount=Decimal("0.00"), created_by=self.user, ) self.SaleItem.objects.create( sale=sale, quantity=Decimal("2"), sale_price=Decimal("1000.00"), ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "1200.00", "apply_to_opening_balance": False, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) sale.refresh_from_db() self.assertEqual(sale.received_amount, Decimal("1200.00")) self.assertEqual(payment.allocations.first().target_type, "SALE") self.assertEqual(payment.allocations.first().target_id, sale.id) # Void restores sale received_amount payment.void(user=self.user) sale.refresh_from_db() self.assertEqual(sale.received_amount, Decimal("0.00")) # Purchase supplier = self.Supplier.objects.create(name="Supp Purchase", shop=self.shop, created_by=self.user) purchase = self.Purchase.objects.create( supplier=supplier, shop=self.shop, currency=self.afn, amount=Decimal("0.00"), received_amount=Decimal("0.00"), created_by=self.user, ) self.PurchaseItem.objects.create( purchase=purchase, quantity=Decimal("3"), buy_price=Decimal("1000.00"), ) s_resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "1500.00", "apply_to_opening_balance": False, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(s_resp.status_code, 201) s_payment = self.SupplierBalancePayment.objects.get(pk=s_resp.data["id"]) purchase.refresh_from_db() self.assertEqual(purchase.received_amount, Decimal("1500.00")) s_payment.void(user=self.user) purchase.refresh_from_db() self.assertEqual(purchase.received_amount, Decimal("0.00")) # 7. Payment allocated to debtor loans def test_07_payment_allocated_to_debtor_loans(self): customer = self.Customer.objects.create(name="Cust Loan", shop=self.shop, created_by=self.user) loan = self.CustomerDebtor.objects.create( customer=customer, currency=self.afn, amount=Decimal("4000.00"), settled_amount=Decimal("0.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "1800.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) loan.refresh_from_db() self.assertEqual(loan.settled_amount, Decimal("1800.00")) self.assertEqual(payment.allocations.first().target_type, "DEBTOR_LOAN") payment.void(user=self.user) loan.refresh_from_db() self.assertEqual(loan.settled_amount, Decimal("0.00")) # 8. Unallocated customer credit and supplier advance def test_08_unallocated_customer_credit_and_supplier_advance(self): customer = self.Customer.objects.create(name="Cust Credit", shop=self.shop, created_by=self.user) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "2500.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) self.assertEqual(payment.unallocated_amount, Decimal("2500.00")) self.assertEqual(customer.get_credit_by_currency(), {"AFN": 2500.0}) payment.void(user=self.user) customer.refresh_from_db() self.assertEqual(customer.get_credit_by_currency(), {}) # Supplier advance supplier = self.Supplier.objects.create(name="Supp Advance", shop=self.shop, created_by=self.user) s_resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "settlement_currency": self.afn.id, "amount": "3500.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(s_resp.status_code, 201) s_payment = self.SupplierBalancePayment.objects.get(pk=s_resp.data["id"]) self.assertEqual(s_payment.unallocated_amount, Decimal("3500.00")) self.assertEqual(supplier.get_credit_by_currency(), {"AFN": 3500.0}) s_payment.void(user=self.user) supplier.refresh_from_db() self.assertEqual(supplier.get_credit_by_currency(), {}) # 9. Duplicate cancellation attempts def test_09_duplicate_cancellation_attempts(self): customer = self.Customer.objects.create(name="Cust Double Cancel", shop=self.shop, created_by=self.user) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) payment_id = resp.data["id"] # Cancel first time first_del = self.client.delete(f"/api/shop/customer-balance-payments/{payment_id}/") self.assertEqual(first_del.status_code, 204) # Cancel second time -> 400 second_del = self.client.delete(f"/api/shop/customer-balance-payments/{payment_id}/") self.assertEqual(second_del.status_code, 400) self.assertIn("already been cancelled", str(second_del.data)) # 10. Decreasing or changing a posted payment def test_10_decreasing_or_changing_a_posted_payment(self): customer = self.Customer.objects.create(name="Cust Edit Test", shop=self.shop, created_by=self.user) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) payment_id = resp.data["id"] # Attempt to reduce amount -> 400 patch_resp = self.client.patch( f"/api/shop/customer-balance-payments/{payment_id}/", {"amount": "500.00"}, format="json", ) self.assertEqual(patch_resp.status_code, 400) self.assertIn("Posted balance payments cannot be modified", str(patch_resp.data)) # Attempt to change currency -> 400 curr_patch = self.client.patch( f"/api/shop/customer-balance-payments/{payment_id}/", {"currency": self.usd.id}, format="json", ) self.assertEqual(curr_patch.status_code, 400) # Updating notes only -> 200 notes_patch = self.client.patch( f"/api/shop/customer-balance-payments/{payment_id}/", {"notes": "Updated audit memo"}, format="json", ) self.assertEqual(notes_patch.status_code, 200) # 11. Historical payments without allocation details def test_11_historical_payments_without_allocation_details(self): customer = self.Customer.objects.create(name="Cust Historical", shop=self.shop, created_by=self.user) # Create directly without allocation rows historical = self.CustomerBalancePayment.objects.create( customer=customer, account=self.afn_account, currency=self.afn, amount=Decimal("5000.00"), allocated_to_afn=Decimal("5000.00"), payment_date=timezone.now(), created_by=self.user, ) from django.core.exceptions import ValidationError as DjangoValidationError with self.assertRaises(DjangoValidationError) as ctx: historical.void(user=self.user) self.assertIn("lacks detailed allocation records", str(ctx.exception)) # 12. Unknown or missing currency migration scenarios def test_12_unknown_or_missing_currency_migration_scenarios(self): self.assertEqual(self.classify_currency_code("AFN"), "AFN") self.assertEqual(self.classify_currency_code("Afghani"), "AFN") self.assertEqual(self.classify_currency_code("USD"), "USD") self.assertEqual(self.classify_currency_code("US Dollar"), "USD") self.assertIsNone(self.classify_currency_code("EUR")) self.assertIsNone(self.classify_currency_code("PKR")) self.assertIsNone(self.classify_currency_code("")) self.assertIsNone(self.classify_currency_code(None)) # 13. Financial transaction and journal integrity def test_13_financial_transaction_and_journal_integrity(self): customer = self.Customer.objects.create(name="Cust Txn Test", shop=self.shop, created_by=self.user) initial_balance = self.afn_account.opening_balance resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "2000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) payment_id = resp.data["id"] txns = self.AccountTransaction.objects.filter( source_type="CUSTOMER_BALANCE_PAYMENT", source_id=str(payment_id) ) self.assertEqual(txns.count(), 2) # Cancel self.client.delete(f"/api/shop/customer-balance-payments/{payment_id}/") # Original transactions must NOT be deleted txns = self.AccountTransaction.objects.filter( source_type="CUSTOMER_BALANCE_PAYMENT", source_id=str(payment_id) ) self.assertEqual(txns.count(), 2) for t in txns: self.assertTrue(t.is_reversed) # Reversal entries must exist reversals = self.AccountTransaction.objects.filter( source_type="REVERSAL", reversed_transaction__in=txns ) self.assertEqual(reversals.count(), 2) # 14. Idempotency and duplicate request prevention def test_14_idempotency_and_duplicate_request_prevention(self): customer = self.Customer.objects.create(name="Cust Idemp Test", shop=self.shop, created_by=self.user) # 1st request with client_reference_id resp1 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "client_reference_id": "IDEMP-TEST-999", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp1.status_code, 201) # Duplicate submission with same client_reference_id -> 400 resp2 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "client_reference_id": "IDEMP-TEST-999", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp2.status_code, 400) self.assertIn("has already been processed", str(resp2.data)) # 15. Insufficient opening balance paid counter during reversal raises ValidationError without silent clamping def test_15_reversal_rejected_on_insufficient_opening_balance_paid(self): from django.core.exceptions import ValidationError as DjangoValidationError customer = self.Customer.objects.create( name="Cust Void Clamp Test", shop=self.shop, opening_balance_afn=Decimal("5000.00"), created_by=self.user, ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "2000.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment_id = resp.data["id"] payment = self.CustomerBalancePayment.objects.get(pk=payment_id) # Artificially alter customer opening_balance_afn_paid to a value less than the allocation customer.refresh_from_db() self.assertEqual(customer.opening_balance_afn_paid, Decimal("2000.00")) customer.opening_balance_afn_paid = Decimal("500.00") # Less than 2000 customer.save(update_fields=["opening_balance_afn_paid"]) # Attempt to void payment with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("is less than allocation amount", str(ctx.exception)) # Verify customer was NOT clamped to 0 customer.refresh_from_db() self.assertEqual(customer.opening_balance_afn_paid, Decimal("500.00")) payment.refresh_from_db() self.assertFalse(payment.is_void) # 16. Insufficient sale / purchase received amount during reversal raises ValidationError def test_16_reversal_rejected_on_insufficient_sale_or_purchase_received_amount(self): from django.core.exceptions import ValidationError as DjangoValidationError # Test on Sale customer = self.Customer.objects.create( name="Cust Sale Recv Clamp Test", shop=self.shop, created_by=self.user ) sale = self.Sale.objects.create( customer=customer, shop=self.shop, currency=self.afn, created_by=self.user, ) self.SaleItem.objects.create( sale=sale, quantity=Decimal("1"), sale_price=Decimal("3000.00"), ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "2000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201) payment_id = resp.data["id"] payment = self.CustomerBalancePayment.objects.get(pk=payment_id) # Alter sale.received_amount to be less than allocation sale.refresh_from_db() self.assertEqual(sale.received_amount, Decimal("2000.00")) sale.received_amount = Decimal("500.00") sale.save(update_fields=["received_amount"]) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("is less than allocation amount", str(ctx.exception)) sale.refresh_from_db() self.assertEqual(sale.received_amount, Decimal("500.00")) payment.refresh_from_db() self.assertFalse(payment.is_void) # Test on Purchase supplier = self.Supplier.objects.create( name="Supp Purch Recv Clamp Test", shop=self.shop, created_by=self.user ) purchase = self.Purchase.objects.create( supplier=supplier, shop=self.shop, currency=self.afn, created_by=self.user, ) self.PurchaseItem.objects.create( purchase=purchase, quantity=Decimal("1"), buy_price=Decimal("4000.00"), ) s_resp = self.client.post( "/api/shop/supplier-balance-payments/", { "supplier": supplier.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "2500.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(s_resp.status_code, 201) s_payment_id = s_resp.data["id"] s_payment = self.SupplierBalancePayment.objects.get(pk=s_payment_id) # Alter purchase.received_amount purchase.refresh_from_db() self.assertEqual(purchase.received_amount, Decimal("2500.00")) purchase.received_amount = Decimal("300.00") purchase.save(update_fields=["received_amount"]) with self.assertRaises(DjangoValidationError) as s_ctx: s_payment.void(user=self.user) self.assertIn("is less than allocation amount", str(s_ctx.exception)) purchase.refresh_from_db() self.assertEqual(purchase.received_amount, Decimal("300.00")) s_payment.refresh_from_db() self.assertFalse(s_payment.is_void) # 17. Reversal rejected on missing or ambiguous accounting transactions def test_17_reversal_rejected_on_missing_or_ambiguous_account_transactions(self): from django.core.exceptions import ValidationError as DjangoValidationError customer = self.Customer.objects.create( name="Cust Txn Ambiguity Test", shop=self.shop, created_by=self.user ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1500.00", "payment_date": timezone.now().isoformat(), }, format="json", ) payment_id = resp.data["id"] payment = self.CustomerBalancePayment.objects.get(pk=payment_id) # Case A: Relink one transaction to another source_id (simulating missing transaction) txns = list(self.AccountTransaction.objects.filter( source_type="CUSTOMER_BALANCE_PAYMENT", source_id=str(payment_id) )) self.assertEqual(len(txns), 2) self.AccountTransaction.objects.filter(pk=txns[0].pk).update(source_id="999999") with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("Expected exactly 2 accounting transactions", str(ctx.exception)) # Restore the missing txn self.AccountTransaction.objects.filter(pk=txns[0].pk).update(source_id=str(payment_id)) # Case B: Create an extra 3rd transaction (ambiguous transactions) self.AccountTransaction.objects.create( account=self.afn_account, account_currency=self.afn, transaction_currency=self.afn, transaction_type=self.AccountTransactionType.CREDIT, amount=Decimal("1500.00"), exchange_rate=Decimal("1"), converted_amount=Decimal("1500.00"), description=f"Duplicate dummy transaction for payment #{payment_id}", transaction_date=timezone.localdate(), source_type="CUSTOMER_BALANCE_PAYMENT", source_id=str(payment_id), is_system_generated=True, ) with self.assertRaises(DjangoValidationError) as ctx2: payment.void(user=self.user) self.assertIn("Expected exactly 2 accounting transactions", str(ctx2.exception)) # 18. Reversal rejected on inconsistent allocations def test_18_reversal_rejected_on_inconsistent_allocations(self): from django.core.exceptions import ValidationError as DjangoValidationError customer1 = self.Customer.objects.create(name="Cust Inconsistent 1", shop=self.shop, created_by=self.user) customer2 = self.Customer.objects.create(name="Cust Inconsistent 2", shop=self.shop, created_by=self.user) sale2 = self.Sale.objects.create(customer=customer2, shop=self.shop, currency=self.afn, created_by=self.user) self.SaleItem.objects.create( sale=sale2, quantity=Decimal("1"), sale_price=Decimal("2000.00") ) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer1.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) payment_id = resp.data["id"] payment = self.CustomerBalancePayment.objects.get(pk=payment_id) # Case A: Tampered allocation target belonging to another customer alloc = self.CustomerBalancePaymentAllocation.objects.create( payment=payment, target_type=self.CustomerBalancePaymentAllocation.TargetType.SALE, target_id=sale2.id, # customer2's sale! target_currency=self.afn, allocated_amount=Decimal("1000.00"), ) payment.allocated_to_afn = Decimal("1000.00") payment.save(update_fields=["allocated_to_afn"]) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("does not belong to customer", str(ctx.exception)) # Fix customer ownership, but tamper currency (Case B) sale1 = self.Sale.objects.create(customer=customer1, shop=self.shop, currency=self.afn, created_by=self.user) self.SaleItem.objects.create( sale=sale1, quantity=Decimal("1"), sale_price=Decimal("2000.00") ) sale1.received_amount = Decimal("1000.00") sale1.save(update_fields=["received_amount"]) alloc.target_id = sale1.id alloc.target_currency = self.usd # Mismatched currency (USD vs AFN sale) alloc.save(update_fields=["target_id", "target_currency"]) payment.allocated_to_afn = Decimal("0.00") payment.allocated_to_usd = Decimal("1000.00") payment.save(update_fields=["allocated_to_afn", "allocated_to_usd"]) with self.assertRaises(DjangoValidationError) as ctx2: payment.void(user=self.user) self.assertIn("currency does not match allocation currency", str(ctx2.exception)) # Case C: Allocation sum totals mismatch payment recorded metrics alloc.target_currency = self.afn alloc.allocated_amount = Decimal("1000.00") alloc.save(update_fields=["target_currency", "allocated_amount"]) payment.allocated_to_afn = Decimal("800.00") # Mismatch (800 vs 1000) payment.allocated_to_usd = Decimal("0.00") payment.save(update_fields=["allocated_to_afn", "allocated_to_usd"]) with self.assertRaises(DjangoValidationError) as ctx3: payment.void(user=self.user) self.assertIn("allocation total for AFN", str(ctx3.exception)) # Case D: One allocation already marked reversed payment.allocated_to_afn = Decimal("1000.00") payment.save(update_fields=["allocated_to_afn"]) alloc.is_reversed = True alloc.save(update_fields=["is_reversed"]) with self.assertRaises(DjangoValidationError) as ctx4: payment.void(user=self.user) self.assertIn("already marked as reversed", str(ctx4.exception)) # 19. Legitimate repeated identical payments are allowed def test_19_legitimate_repeated_identical_payments_allowed(self): customer = self.Customer.objects.create(name="Cust Repeated Pay Test", shop=self.shop, created_by=self.user) now_str = timezone.now().isoformat() # Submit first payment resp1 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "500.00", "payment_date": now_str, }, format="json", ) self.assertEqual(resp1.status_code, 201) # Submit second identical payment immediately (same amount, customer, account, date) -> MUST SUCCEED (201) resp2 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "500.00", "payment_date": now_str, }, format="json", ) self.assertEqual(resp2.status_code, 201) self.assertNotEqual(resp1.data["id"], resp2.data["id"]) # With explicit client_reference_id: resp_idemp1 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "500.00", "client_reference_id": "UNIQUE-TXN-001", "payment_date": now_str, }, format="json", ) self.assertEqual(resp_idemp1.status_code, 201) # Duplicate submission with same client_reference_id -> MUST BE REJECTED (400) resp_idemp2 = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "500.00", "client_reference_id": "UNIQUE-TXN-001", "payment_date": now_str, }, format="json", ) self.assertEqual(resp_idemp2.status_code, 400) self.assertIn("has already been processed", str(resp_idemp2.data)) # 20. Forward audit unknown currencies migration function test def test_20_forward_audit_unknown_currencies_migration(self): from shop_management.models.data.utils import Currency import importlib mig_mod = importlib.import_module("shop_management.migrations.0111_balance_payment_allocations_and_void") forward_audit_unknown_currencies = mig_mod.forward_audit_unknown_currencies from django.apps import apps eur = Currency.objects.create(title="EUR", description="Euro", shop=self.shop) cust_ambig = self.Customer.objects.create( name="Ambig EUR Cust", shop=self.shop, opening_balance=Decimal("3000.00"), opening_balance_currency=eur, created_by=self.user, ) supp_ambig = self.Supplier.objects.create( name="Ambig EUR Supp", shop=self.shop, opening_balance=Decimal("4000.00"), opening_balance_currency=eur, created_by=self.user, ) result = forward_audit_unknown_currencies(apps, None) self.assertIn("ambiguous_customers", result) self.assertIn("ambiguous_suppliers", result) flagged_cust_ids = [c["id"] for c in result["ambiguous_customers"]] flagged_supp_ids = [s["id"] for s in result["ambiguous_suppliers"]] self.assertIn(cust_ambig.id, flagged_cust_ids) self.assertIn(supp_ambig.id, flagged_supp_ids) # Verify read-only safety: balances and currencies were NOT altered cust_ambig.refresh_from_db() self.assertEqual(cust_ambig.opening_balance, Decimal("3000.00")) self.assertEqual(cust_ambig.opening_balance_currency_id, eur.id) supp_ambig.refresh_from_db() self.assertEqual(supp_ambig.opening_balance, Decimal("4000.00")) self.assertEqual(supp_ambig.opening_balance_currency_id, eur.id) # ---- Final targeted integrity corrections (tests 21-25) ---- def _make_opening_payment(self, kind, name): """Create a payment that settles a 2000 AFN opening balance; returns (party, payment).""" from django.core.exceptions import ValidationError as DjangoValidationError # noqa: F401 if kind == "customer": party = self.Customer.objects.create( name=name, shop=self.shop, opening_balance_afn=Decimal("5000.00"), created_by=self.user ) url, key, model = ( "/api/shop/customer-balance-payments/", "customer", self.CustomerBalancePayment, ) else: party = self.Supplier.objects.create( name=name, shop=self.shop, opening_balance_afn=Decimal("5000.00"), created_by=self.user ) url, key, model = ( "/api/shop/supplier-balance-payments/", "supplier", self.SupplierBalancePayment, ) resp = self.client.post( url, { key: party.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "2000.00", "apply_to_opening_balance": True, "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201, resp.data) payment = model.objects.get(pk=resp.data["id"]) self.assertEqual(payment.allocations.count(), 1) return party, payment def _assert_nothing_changed(self, kind, party, payment): party.refresh_from_db() payment.refresh_from_db() self.assertFalse(payment.is_void) self.assertEqual(party.opening_balance_afn_paid, Decimal("2000.00")) self.assertFalse(payment.allocations.filter(is_reversed=True).exists()) src = "CUSTOMER_BALANCE_PAYMENT" if kind == "customer" else "SUPPLIER_BALANCE_PAYMENT" self.assertFalse(self.AccountTransaction.objects.filter(source_type="REVERSAL").exists()) self.assertFalse( self.AccountTransaction.objects.filter(source_type=src, reversed_at__isnull=False).exists() ) # 21. Unsupported allocation target types are rejected, never marked reversed def test_21_reversal_rejected_on_unsupported_allocation_target_type(self): from django.core.exceptions import ValidationError as DjangoValidationError for kind in ("customer", "supplier"): party, payment = self._make_opening_payment(kind, f"Unsupported Type {kind}") alloc_model = ( self.CustomerBalancePaymentAllocation if kind == "customer" else self.SupplierBalancePaymentAllocation ) alloc_model.objects.filter(payment_id=payment.id).update(target_type="MYSTERY_TARGET") with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("unsupported target type", str(ctx.exception)) self._assert_nothing_changed(kind, party, payment) # Cross-kind type (a PURCHASE allocation on a customer payment) is also unsupported party, payment = self._make_opening_payment("customer", "Cross Kind Type") self.CustomerBalancePaymentAllocation.objects.filter(payment_id=payment.id).update( target_type="PURCHASE" ) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("unsupported target type", str(ctx.exception)) self._assert_nothing_changed("customer", party, payment) # 22. Two transactions with the same direction are rejected (exact and historical matching) def test_22_reversal_rejected_on_same_direction_transactions(self): from django.core.exceptions import ValidationError as DjangoValidationError for kind in ("customer", "supplier"): src = "CUSTOMER_BALANCE_PAYMENT" if kind == "customer" else "SUPPLIER_BALANCE_PAYMENT" party, payment = self._make_opening_payment(kind, f"Same Direction {kind}") pay_txn = self.AccountTransaction.objects.get( source_type=src, source_id=str(payment.id), account=self.afn_account ) clearing_txn = self.AccountTransaction.objects.exclude(pk=pay_txn.pk).get( source_type=src, source_id=str(payment.id) ) self.AccountTransaction.objects.filter(pk=clearing_txn.pk).update( transaction_type=pay_txn.transaction_type ) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("unexpected direction", str(ctx.exception)) self._assert_nothing_changed(kind, party, payment) # Historical (description-based) matching gets identical validation self.AccountTransaction.objects.filter( source_type=src, source_id=str(payment.id) ).update(source_id="") with self.assertRaises(DjangoValidationError) as ctx2: payment.void(user=self.user) self.assertIn("unexpected direction", str(ctx2.exception)) self._assert_nothing_changed(kind, party, payment) # 23. Incorrect clearing account is rejected (exact and historical matching) def test_23_reversal_rejected_on_incorrect_clearing_account(self): from django.core.exceptions import ValidationError as DjangoValidationError from shop_management.models.data.shop_account import Account rogue = Account.objects.create( shop=self.shop, name="Rogue AFN Account", currency=self.afn, opening_balance=Decimal("0.00"), allow_overdraft=True, ) for kind in ("customer", "supplier"): src = "CUSTOMER_BALANCE_PAYMENT" if kind == "customer" else "SUPPLIER_BALANCE_PAYMENT" party, payment = self._make_opening_payment(kind, f"Wrong Clearing {kind}") pay_txn = self.AccountTransaction.objects.get( source_type=src, source_id=str(payment.id), account=self.afn_account ) clearing_txn = self.AccountTransaction.objects.exclude(pk=pay_txn.pk).get( source_type=src, source_id=str(payment.id) ) self.AccountTransaction.objects.filter(pk=clearing_txn.pk).update(account=rogue) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("expected clearing account", str(ctx.exception)) self._assert_nothing_changed(kind, party, payment) self.AccountTransaction.objects.filter( source_type=src, source_id=str(payment.id) ).update(source_id="") with self.assertRaises(DjangoValidationError) as ctx2: payment.void(user=self.user) self.assertIn("expected clearing account", str(ctx2.exception)) self._assert_nothing_changed(kind, party, payment) # 24. Valid historical (description-matched) payments still cancel correctly def test_24_historical_description_matched_payment_can_be_voided(self): for kind in ("customer", "supplier"): src = "CUSTOMER_BALANCE_PAYMENT" if kind == "customer" else "SUPPLIER_BALANCE_PAYMENT" party, payment = self._make_opening_payment(kind, f"Historical OK {kind}") self.AccountTransaction.objects.filter( source_type=src, source_id=str(payment.id) ).update(source_id="") payment.void(user=self.user, reason="test") payment.refresh_from_db() party.refresh_from_db() self.assertTrue(payment.is_void) self.assertEqual(party.opening_balance_afn_paid, Decimal("0.00")) self.assertEqual( self.AccountTransaction.objects.filter(source_type="REVERSAL").count() % 2, 0 ) # 25. Credit-application guard covers draft and posted sales, but not cancelled ones def test_25_credit_application_guard_covers_draft_and_posted_sales(self): from django.core.exceptions import ValidationError as DjangoValidationError from shop_management.models.data.customer_credit_application import CustomerCreditApplication customer = self.Customer.objects.create(name="Credit Guard", shop=self.shop, created_by=self.user) resp = self.client.post( "/api/shop/customer-balance-payments/", { "customer": customer.id, "account": self.afn_account.id, "currency": self.afn.id, "amount": "1000.00", "payment_date": timezone.now().isoformat(), }, format="json", ) self.assertEqual(resp.status_code, 201, resp.data) payment = self.CustomerBalancePayment.objects.get(pk=resp.data["id"]) self.assertEqual(payment.unallocated_amount, Decimal("1000.00")) sale = self.Sale.objects.create( customer=customer, shop=self.shop, currency=self.afn, uses_inventory_engine=True, created_by=self.user, ) self.assertEqual(sale.document_status, "DRAFT") CustomerCreditApplication.objects.create( customer=customer, sale=sale, balance_payment=payment, source_currency=self.afn, sale_currency=self.afn, source_amount=Decimal("500.00"), sale_amount=Decimal("500.00"), created_by=self.user, ) for status in ("DRAFT", "POSTED"): self.Sale.objects.filter(pk=sale.pk).update(document_status=status) with self.assertRaises(DjangoValidationError) as ctx: payment.void(user=self.user) self.assertIn("applied to a sale", str(ctx.exception)) payment.refresh_from_db() self.assertFalse(payment.is_void) # A cancelled sale no longer consumes the credit, so cancellation is allowed self.Sale.objects.filter(pk=sale.pk).update(document_status="VOID") payment.void(user=self.user, reason="ok") payment.refresh_from_db() self.assertTrue(payment.is_void)