import copy from django.contrib.auth import get_user_model from django.test import TestCase, override_settings from rest_framework import status from rest_framework.test import APIClient from basic_info.models import ( Merchant, MerchantTypeEnum, Customer, WareHouse, WareHouseModeEnum, ProductCategory, Product, ProductUnitEnum, Employee, EmployeeStatusEnum, ) from business import models as business_models from stock import models as stock_models from business import allocation_services from stock import models as stock_models @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class PreSalesOrderAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='预销售商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='预销售客户', created_by=None, ) self.warehouse = WareHouse.objects.create( merchant=self.merchant, name='预销售仓库', mode=WareHouseModeEnum.UNRESTRICTED, ) self.warehouse_strict_out = WareHouse.objects.create( merchant=self.merchant, name='预销售严出仓', mode=WareHouseModeEnum.RESTRICT_IN_OUT, ) category = ProductCategory.objects.create( merchant=self.merchant, name='预销售品类', product_prefix='PRE', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='预销售产品', human_id='PRE-001', unit=ProductUnitEnum.METER, ) User = get_user_model() self.user = User.objects.create_user(username='pre_sales_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='预销售员', status=EmployeeStatusEnum.ACTIVE, ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.payload = { 'customer': self.customer.id, 'warehouse': self.warehouse.id, 'kind': business_models.SalesOrderKindEnum.WHOLESALE, 'remarks': '预销售单备注', 'items': [ { 'product_id': self.product.id, 'quantity': '12.5', 'unit': '米', 'order_quantity': 7, 'remarks': '明细备注', } ], } self.strict_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_strict_out.id, 'kind': business_models.SalesOrderKindEnum.WHOLESALE, 'remarks': '预销售单备注', 'items': [ { 'product_id': self.product.id, 'quantity': '10', 'unit': '米', } ], } def _create(self, payload=None) -> int: resp = self.client.post('/api/v1/pre-sales-orders/', payload or self.payload, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) return resp.data['id'] def test_create_pre_sales_order_success(self): resp = self.client.post('/api/v1/pre-sales-orders/', self.payload, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) self.assertIn('id', resp.data) self.assertIn('human_id', resp.data) self.assertEqual(resp.data['customer'], self.customer.id) self.assertEqual(resp.data['warehouse'], self.warehouse.id) self.assertEqual(len(resp.data['items']), 1) def test_list_pre_sales_orders(self): self._create() resp = self.client.get('/api/v1/pre-sales-orders/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIn('results', resp.data) self.assertGreaterEqual(len(resp.data['results']), 1) def test_get_detail(self): order_id = self._create() resp = self.client.get(f'/api/v1/pre-sales-orders/{order_id}/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['id'], order_id) self.assertIn('items', resp.data) def test_get_detail_includes_allocation_records_with_stock_details(self): resp = self.client.post('/api/v1/pre-sales-orders/', self.payload, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) order_id = resp.data['id'] item_id = resp.data['items'][0]['id'] stock_record = stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse, source_type=stock_models.StockChangeSourceEnum.PURCHASE, created_by=self.user, ) stock_detail = stock_models.StockChangeDetail.objects.create( merchant=self.merchant, product=self.product, unit=ProductUnitEnum.METER, stock_change_record=stock_record, quantity='12.5', ) create_resp = self.client.post( f'/api/v1/pre-sales-order-items/{item_id}/allocations/', { 'stock_ids': [stock_detail.id], 'quantity': '5.5', 'unit': '米', }, format='json', ) self.assertEqual(create_resp.status_code, status.HTTP_201_CREATED) detail_resp = self.client.get(f'/api/v1/pre-sales-orders/{order_id}/') self.assertEqual(detail_resp.status_code, status.HTTP_200_OK) records = detail_resp.data['items'][0]['allocation_records'] self.assertEqual(records[0]['stock_ids'], [stock_detail.id]) self.assertEqual(records[0]['stock_details'][0]['product_id'], self.product.id) self.assertEqual(records[0]['stock_details'][0]['product_name'], self.product.name) def test_update_pre_sales_order(self): order_id = self._create() payload = copy.deepcopy(self.payload) payload['remarks'] = '更新备注' payload['items'][0]['quantity'] = '20' resp = self.client.put(f'/api/v1/pre-sales-orders/{order_id}/', payload, format='json') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['remarks'], '更新备注') self.assertEqual(str(resp.data['items'][0]['quantity']), '20.00') def test_delete_pre_sales_order(self): order_id = self._create() resp = self.client.delete(f'/api/v1/pre-sales-orders/{order_id}/') self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) self.assertFalse(business_models.PreSalesOrder.objects.filter(id=order_id).exists()) def test_convert_to_sales_order_rejects_non_strict_out(self): order_id = self._create() resp = self.client.post(f'/api/v1/pre-sales-orders/{order_id}/convert-to-sales/', {}, format='json') self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('仅支持严进严出仓库模式', resp.data.get('error', '')) def test_convert_to_sales_order_success(self): resp = self.client.post('/api/v1/pre-sales-orders/', self.strict_payload, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) order_id = resp.data['id'] order = business_models.PreSalesOrder.objects.get(id=order_id) item = order.items.first() stock_record = stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_strict_out, source_type=stock_models.StockChangeSourceEnum.PURCHASE, created_by=self.user, ) stock_detail = stock_models.StockChangeDetail.objects.create( merchant=self.merchant, product=self.product, unit=ProductUnitEnum.METER, stock_change_record=stock_record, quantity='10', ) allocation_services.create_allocation_record( item=item, stock_ids=[stock_detail.id], quantity='10', unit='米', scanned_by=self.employee, ) convert_resp = self.client.post( f'/api/v1/pre-sales-orders/{order_id}/convert-to-sales/', {}, format='json', ) self.assertEqual(convert_resp.status_code, status.HTTP_201_CREATED) self.assertIn('id', convert_resp.data) self.assertEqual(convert_resp.data['status'], business_models.SalesOrderStatusEnum.PENDING) sales_order = business_models.SalesOrder.objects.get(id=convert_resp.data['id']) self.assertEqual(sales_order.from_pre_sales_order_id, order_id)