from datetime import timedelta from decimal import Decimal from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from shop_management.models.data.shop import Shop from shop_management.models.data.organization import Warehouse from shop_management.models.data.utils import Currency, Unit from shop_management.models.data.product import Product from shop_management.models.data.sale import Sale, SaleItem from shop_management.models.data.sale_return import SaleReturn, SaleReturnItem, SaleReturnItemBatchRestoration from shop_management.models.data.inventory import InventoryBatch, StockMovement class WarehouseProfitTests(TestCase): def setUp(self): self.user = get_user_model().objects.create_user(username='warehouse-profit', password='test', is_staff=True) self.user.allowed_tabs = ['reports'] self.user.save(update_fields=['allowed_tabs']) self.shop = Shop.objects.create(owner=self.user, name='Kabul') self.first = Warehouse.objects.create(shop=self.shop, code='A', name='Warehouse A') self.second = Warehouse.objects.create(shop=self.shop, code='B', name='Warehouse B') self.currency = Currency.objects.create(shop=self.shop, title='AFN') self.usd = Currency.objects.create(shop=self.shop, title='USD') unit = Unit.objects.create(shop=self.shop, title='Each') self.product = Product.objects.create(shop=self.shop, name='Medicine', default_unit=unit) self.batch = InventoryBatch.objects.create(product=self.product, batch_number='TEST') self.today = timezone.localdate() self.client = APIClient() self.client.force_authenticate(self.user) def sale(self, warehouse=None, currency=None, total='90', cost='60', day=None, status='POSTED'): sale = Sale.objects.create(shop=self.shop, warehouse=warehouse or self.first, currency=currency or self.currency, invoice_date=day or self.today, uses_inventory_engine=True, document_status=status, calculated_subtotal=100, calculated_total=Decimal(total), discount_amount=10) SaleItem.objects.create(sale=sale, product=self.product, unit=self.product.default_unit, quantity=2, sale_price=50, cogs_total=cost) return sale def returned(self, sale, amount='45', cost='30', status='POSTED'): document = SaleReturn.objects.create(shop=self.shop, sale=sale, uses_inventory_engine=True, document_status=status, total_refound_amount=amount) item = SaleReturnItem.objects.create(sale_return=document, sale_item=sale.sale_items.get(), return_quantity=1, refund_amount=amount) SaleReturnItemBatchRestoration.objects.create(return_item=item, batch=self.batch, quantity=1, unit_cost=cost) return document def report(self, **filters): return self.client.get('/api/shop/warehouse-profit-report/', {'shop':self.shop.pk, 'start_date':str(self.today), 'end_date':str(self.today), **filters}) def test_profit_is_grouped_by_warehouse_and_currency_and_discounted(self): sale = self.sale() self.returned(sale) self.sale(warehouse=self.second, total='200', cost='100') self.sale(currency=self.usd, total='10', cost='6') self.sale(status='DRAFT', total='999') self.sale(status='VOID', total='999') response = self.report() self.assertEqual(response.status_code, 200, response.data) self.assertEqual(len(response.data['rows']), 3) row = next(row for row in response.data['rows'] if row['warehouse_id']==self.first.pk and row['currency']=='AFN') self.assertEqual(row['sales'], '90.00') self.assertEqual(row['net_sales'], '45.00') self.assertEqual(row['net_cogs'], '30.00') self.assertEqual(row['gross_profit'], '15.00') self.assertEqual(row['margin_percent'], '33.33') summaries = {row['currency']:row for row in response.data['summary_by_currency']} self.assertEqual(summaries['AFN']['gross_profit'], '115.00') self.assertEqual(summaries['USD']['gross_profit'], '4.00') def test_mixed_warehouse_invoice_is_allocated_by_its_lines(self): sale = Sale.objects.create( shop=self.shop, warehouse=self.first, currency=self.currency, invoice_date=self.today, uses_inventory_engine=True, document_status=Sale.DocumentStatus.POSTED, calculated_subtotal=Decimal("200"), calculated_total=Decimal("180"), discount_amount=Decimal("20"), ) SaleItem.objects.create( sale=sale, warehouse=self.first, product=self.product, unit=self.product.default_unit, quantity=1, sale_price=Decimal("50"), cogs_total=Decimal("30"), ) SaleItem.objects.create( sale=sale, warehouse=self.second, product=self.product, unit=self.product.default_unit, quantity=3, sale_price=Decimal("50"), cogs_total=Decimal("90"), ) response = self.report() self.assertEqual(response.status_code, 200, response.data) rows = {row["warehouse_id"]: row for row in response.data["rows"]} self.assertEqual(rows[self.first.pk]["sales"], "45.00") self.assertEqual(rows[self.first.pk]["net_cogs"], "30.00") self.assertEqual(rows[self.second.pk]["sales"], "135.00") self.assertEqual(rows[self.second.pk]["net_cogs"], "90.00") self.assertEqual(rows[self.first.pk]["invoice_count"], 1) self.assertEqual(rows[self.second.pk]["invoice_count"], 1) def test_returns_from_an_older_sale_are_counted_in_the_return_period(self): sale = self.sale(day=self.today-timedelta(days=60)) self.returned(sale) response = self.report() row = response.data['rows'][0] self.assertEqual(row['invoice_count'], 0) self.assertEqual(row['return_count'], 1) self.assertEqual(row['net_sales'], '-45.00') self.assertEqual(row['net_cogs'], '-30.00') self.assertEqual(row['gross_profit'], '-15.00') self.assertIsNone(row['margin_percent']) def test_warehouse_filter_and_date_validation(self): self.sale(); self.sale(warehouse=self.second) response = self.report(warehouse=self.second.pk) self.assertEqual(len(response.data['rows']), 1) self.assertEqual(response.data['rows'][0]['warehouse_id'], self.second.pk) self.assertEqual(self.report(start_date='invalid').status_code,400) self.assertEqual(self.report(start_date=str(self.today+timedelta(days=1))).status_code,400) self.assertEqual(self.report(warehouse='invalid').status_code,400) def test_branch_access_is_enforced(self): other = get_user_model().objects.create_user(username='other-profit') branch = Shop.objects.create(owner=other,name='Other') foreign = Warehouse.objects.create(shop=branch,code='MAIN',name='Other Warehouse') self.assertEqual(self.report(shop=branch.pk).status_code,403) self.assertEqual(self.report(warehouse=foreign.pk).status_code,400) scoped = self.client.get("/api/shop/warehouse-profit-report/", {"shop": self.shop.pk}, HTTP_X_SELECTED_SHOP=str(branch.pk)) self.assertEqual(scoped.status_code, 403) def test_draft_and_cancelled_returns_are_not_included(self): sale = self.sale() self.returned(sale,status='DRAFT'); self.returned(sale,status='VOID') row = self.report().data['rows'][0] self.assertEqual(row['returns'],'0.00') self.assertEqual(row['gross_profit'],'30.00') def test_cross_currency_costs_are_flagged_instead_of_misreported(self): sale = self.sale() StockMovement.objects.create(warehouse=self.first, product=self.product, batch=self.batch, movement_type='SALE',quantity_out=2,unit_cost=30,currency=self.usd,reference_type='Sale',reference_id=str(sale.pk)) response = self.report() self.assertTrue(response.data['incomplete_costs']) self.assertIsNone(response.data['rows'][0]['gross_profit']) self.assertIsNone(response.data['summary_by_currency'][0]['gross_profit']) def test_empty_report_and_permission(self): response = self.report() self.assertEqual(response.status_code,200) self.assertEqual(response.data['rows'],[]) self.user.allowed_tabs=[]; self.user.save(update_fields=['allowed_tabs']) self.assertEqual(self.report().status_code,403) def test_all_dates_includes_older_sales_and_explains_inactive_branches(self): self.sale(day=self.today-timedelta(days=60)) self.sale(status='VOID') self.sale(status='DRAFT') recent = self.report() self.assertEqual(recent.data['rows'], []) self.assertEqual(recent.data['activity']['posted'], 1) self.assertEqual(recent.data['activity']['cancelled'], 1) self.assertEqual(recent.data['activity']['drafts'], 1) complete = self.report(period='all') self.assertEqual(complete.status_code, 200, complete.data) self.assertEqual(complete.data['period'], 'all') self.assertEqual(complete.data['rows'][0]['sales'], '90.00') self.assertEqual(complete.data['rows'][0]['gross_profit'], '30.00')