import copy import shutil import tempfile import datetime from pathlib import Path from unittest.mock import patch from decimal import Decimal from django.test import TestCase, override_settings from django.contrib.auth.models import User, Permission from rest_framework.test import APIClient from rest_framework import status from basic_info.models import ( BankAccount, Customer, Employee, EmployeeStatusEnum, Merchant, MerchantTypeEnum, MerchantSetting, MerchantSettingKeyEnum, Product, ProductCategory, ProductUnitEnum, Supplier, UserProfile, WareHouse, WareHouseModeEnum, ) from business import models as business_models, services from stock import models as stock_models from stock import services as stock_services from api_v1 import tasks from printing import models as printing_models class UserCreationAPITestCase(TestCase): """测试用户创建 API""" def setUp(self): """设置测试数据""" # 创建商户 self.merchant = Merchant.objects.create( name='测试商户', type=1 # 假设1是有效的MerchantType ) # 创建管理员用户 self.admin_user = User.objects.create_user( username='admin', password='adminpass123', is_staff=True ) # 授予创建用户所需的权限 add_user_perm = Permission.objects.get(codename='add_user') self.admin_user.user_permissions.add(add_user_perm) self.admin_user.save() # 设置 API 客户端 self.client = APIClient() self.client.force_authenticate(user=self.admin_user) def test_create_user_with_profile(self): """测试创建用户和用户资料""" data = { 'username': 'testuser', 'email': 'test@example.com', 'password': 'testpass123', 'is_staff': False, 'description': '测试用户资料', 'merchant_id': self.merchant.id } response = self.client.post('/api/v1/users/create/', data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # 验证返回的数据 self.assertIn('user', response.data) self.assertIn('profile', response.data) self.assertEqual(response.data['user']['username'], 'testuser') self.assertEqual(response.data['user']['email'], 'test@example.com') self.assertEqual(response.data['user']['is_staff'], False) self.assertEqual(response.data['profile']['description'], '测试用户资料') self.assertEqual(response.data['profile']['merchant'], self.merchant.id) # 验证数据库中的用户 user = User.objects.get(username='testuser') self.assertEqual(user.email, 'test@example.com') self.assertFalse(user.is_staff) # 验证数据库中的用户资料 profile = UserProfile.objects.get(user=user) self.assertEqual(profile.merchant, self.merchant) self.assertEqual(profile.description, '测试用户资料') def test_create_user_with_duplicate_username(self): """测试创建用户时使用重复的用户名""" # 先创建一个用户 User.objects.create_user(username='existinguser', password='pass123') data = { 'username': 'existinguser', # 重复的用户名 'password': 'testpass123', 'merchant_id': self.merchant.id } response = self.client.post('/api/v1/users/create/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('username', response.data) def test_create_user_with_invalid_merchant(self): """测试创建用户时使用无效的merchant_id""" data = { 'username': 'testuser', 'password': 'testpass123', 'merchant_id': 999 # 不存在的merchant_id } response = self.client.post('/api/v1/users/create/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('merchant_id', response.data) def test_create_user_unauthenticated(self): """测试未认证用户创建用户""" self.client.force_authenticate(user=None) data = { 'username': 'testuser', 'password': 'testpass123', 'merchant_id': self.merchant.id } response = self.client.post('/api/v1/users/create/', data, format='json') self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_create_user_with_short_password(self): """测试创建用户时密码过短""" data = { 'username': 'testuser', 'password': '123', # 密码过短 'merchant_id': self.merchant.id } response = self.client.post('/api/v1/users/create/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('password', response.data) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class PurchaseOrderAPITestCase(TestCase): """采购单 API 测试""" def setUp(self): self.merchant = Merchant.objects.create(name='PO商户', type=MerchantTypeEnum.FACTORY) self.supplier = Supplier.objects.create(merchant=self.merchant, name='供应商A') self.warehouse_strict = WareHouse.objects.create( merchant=self.merchant, name='严进仓', mode=WareHouseModeEnum.RESTRICT_IN, ) self.warehouse_relaxed = WareHouse.objects.create( merchant=self.merchant, name='宽进仓', mode=WareHouseModeEnum.UNRESTRICTED, ) category = ProductCategory.objects.create( merchant=self.merchant, name='品类', product_prefix='FAB', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='产品1', human_id='FAB-001', unit=ProductUnitEnum.METER, ) self.user = User.objects.create_user(username='po_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='仓管', ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.strict_payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_strict.id, 'order_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'numbers': [10, 5], 'price': '12.5', 'unit': '米', } ], 'remarks': '接口测试', } self.relaxed_payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_relaxed.id, 'order_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'quantity': 120, 'num_of_rolls': 3, 'price': '10.5', } ], } def _create_purchase_order(self, payload): body = copy.deepcopy(payload) with patch('business.services.create_purchase_order_stock_entries.delay'): response = self.client.post('/api/v1/purchase-orders/', body, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) return response.data['id'] def test_create_purchase_order_success_strict(self): with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay: response = self.client.post('/api/v1/purchase-orders/', self.strict_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) self.assertIn('human_id', response.data) self.assertEqual(response.data['status'], 1) self.assertIn('等待审批', response.data['message']) order = business_models.PurchaseOrder.objects.get(id=response.data['id']) expected_human_id = f"CG{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) mock_delay.assert_not_called() def test_create_purchase_order_with_pre_purchase_binding(self): pre_order = business_models.PrePurchaseOrder.objects.create( merchant=self.merchant, supplier=self.supplier, warehouse=self.warehouse_strict, operator=self.employee, created_by=self.user, ) payload = {**self.strict_payload, 'from_pre_purchase_order_id': pre_order.id} with patch('business.services.create_purchase_order_stock_entries.delay'): response = self.client.post('/api/v1/purchase-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) order = business_models.PurchaseOrder.objects.get(id=response.data['id']) self.assertEqual(order.from_pre_purchase_order_id, pre_order.id) def test_bind_purchase_order_stock_change_record(self): MerchantSetting.objects.filter( merchant=self.merchant, key=MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS, ).update(val_bool=False) order_id = self._create_purchase_order(self.strict_payload) record = stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_strict, source_type=stock_models.StockChangeSourceEnum.PURCHASE, created_by=self.user, ) resp = self.client.post( f'/api/v1/purchase-orders/{order_id}/bind-stock-change/', {'stock_change_record_id': record.id}, format='json', ) self.assertEqual(resp.status_code, status.HTTP_200_OK) record.refresh_from_db() self.assertEqual(record.source_id, order_id) def test_create_purchase_order_invalid_supplier(self): payload = {**self.strict_payload, 'supplier': 999} response = self.client.post('/api/v1/purchase-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不存在', response.data['error']) def test_create_purchase_order_unauthenticated(self): self.client.force_authenticate(user=None) response = self.client.post('/api/v1/purchase-orders/', self.strict_payload, format='json') self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_create_purchase_order_relaxed_mode(self): with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay: response = self.client.post('/api/v1/purchase-orders/', self.relaxed_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('human_id', response.data) self.assertEqual(response.data['status'], 1) order = business_models.PurchaseOrder.objects.get(id=response.data['id']) expected_human_id = f"CG{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) mock_delay.assert_not_called() def test_mode_mismatch_raises(self): payload = {**self.relaxed_payload} payload['warehouse'] = self.warehouse_strict.id # 严进仓却传宽进参数 response = self.client.post('/api/v1/purchase-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('numbers', response.data['error']) def test_review_purchase_order_requires_action(self): order_id = self._create_purchase_order(self.strict_payload) response = self.client.post(f'/api/v1/purchase-orders/{order_id}/review/', {}, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('action', response.data) def test_review_purchase_order_approve_success(self): order_id = self._create_purchase_order(self.strict_payload) response = self.client.post( f'/api/v1/purchase-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.PurchaseOrderStatusEnum.APPROVED) order = business_models.PurchaseOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.PurchaseOrderStatusEnum.APPROVED) def test_review_purchase_order_cancel_blocked_after_stock_exists(self): order_id = self._create_purchase_order(self.relaxed_payload) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_relaxed, source_type=stock_models.StockChangeSourceEnum.PURCHASE, source_id=order_id, ) response = self.client.post( f'/api/v1/purchase-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) order = business_models.PurchaseOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.PurchaseOrderStatusEnum.PENDING) def test_update_purchase_order_success(self): order_id = self._create_purchase_order(self.strict_payload) pre_order = business_models.PrePurchaseOrder.objects.create( merchant=self.merchant, supplier=self.supplier, warehouse=self.warehouse_relaxed, operator=self.employee, created_by=self.user, ) payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_relaxed.id, 'order_date': '2025-12-01', 'items': [ { 'product_id': self.product.id, 'quantity': 200, 'num_of_rolls': 4, 'price': '12.88', } ], 'remarks': '更新采购单', 'from_pre_purchase_order_id': pre_order.id, } response = self.client.put(f'/api/v1/purchase-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['remarks'], '更新采购单') updated_order = business_models.PurchaseOrder.objects.get(id=order_id) self.assertEqual(updated_order.from_pre_purchase_order_id, pre_order.id) def test_update_purchase_order_rejects_non_pending(self): order_id = self._create_purchase_order(self.relaxed_payload) self.client.post( f'/api/v1/purchase-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) payload = copy.deepcopy(self.relaxed_payload) response = self.client.put(f'/api/v1/purchase-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不可修改', response.data['error']) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class SalesOrderAPITestCase(TestCase): """销售单 API 测试""" def setUp(self): self.merchant = Merchant.objects.create(name='销售商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='客户A', mobile='13800000000', created_by=None, ) self.warehouse_strict = WareHouse.objects.create( merchant=self.merchant, name='销售严进仓', mode=WareHouseModeEnum.RESTRICT_IN, ) self.warehouse_relaxed = 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='SAL', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='销售产品1', human_id='SAL-001', unit=ProductUnitEnum.METER, ) self.printing_order_for_sales = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉布', width='150cm', ) self.printing_job_for_sales = printing_models.PrintingJob.objects.create( printing_order=self.printing_order_for_sales, product=self.product, quantity=50, unit='米', ) self.user = User.objects.create_user(username='sales_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='销售员', ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.strict_in_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_strict.id, 'order_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'numbers': [8, 4], 'price': '15.0', 'unit': '米', } ], 'remarks': '销售接口测试', } self.strict_out_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_strict_out.id, 'order_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'consume_detail_ids': [101, 102], 'quantity': 30, 'price': '18.5', 'unit': '米', } ], } self.relaxed_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'order_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'quantity': 90, 'num_of_rolls': 3, 'price': '16.5', } ], } def _create_sales_order(self, payload): body = copy.deepcopy(payload) with patch('business.services.create_sales_order_stock_entries.delay'): response = self.client.post('/api/v1/sales-orders/', body, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) return response.data['id'] def test_create_sales_order_success_strict(self): with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay: response = self.client.post('/api/v1/sales-orders/', self.strict_in_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) self.assertIn('human_id', response.data) order = business_models.SalesOrder.objects.get(id=response.data['id']) expected_human_id = f"XS{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) self.assertEqual(response.data['status'], business_models.SalesOrderStatusEnum.PENDING) mock_delay.assert_not_called() def test_create_sales_order_invalid_customer(self): payload = {**self.strict_in_payload, 'customer': 999} response = self.client.post('/api/v1/sales-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不存在', response.data['error']) def test_create_sales_order_relaxed_mode(self): with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay: response = self.client.post('/api/v1/sales-orders/', self.relaxed_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data['status'], business_models.SalesOrderStatusEnum.PENDING) mock_delay.assert_not_called() def test_create_sales_order_strict_out_success(self): with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay: response = self.client.post('/api/v1/sales-orders/', self.strict_out_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data['status'], business_models.SalesOrderStatusEnum.PENDING) mock_delay.assert_not_called() def test_create_sales_order_strict_out_requires_consume_ids(self): payload = copy.deepcopy(self.strict_out_payload) payload['items'][0].pop('consume_detail_ids') response = self.client.post('/api/v1/sales-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('consume_detail_ids', response.data['error']) def test_update_sales_order_success(self): order_id = self._create_sales_order(self.strict_in_payload) payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'order_date': '2025-12-01', 'items': [ { 'product_id': self.product.id, 'quantity': 100, 'num_of_rolls': 2, 'price': '22.5', } ], 'remarks': '更新后的销售单', } response = self.client.put(f'/api/v1/sales-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['remarks'], '更新后的销售单') self.assertEqual(response.data['quantity_of_rolls'], [[]]) def test_get_sales_order_detail_includes_human_id(self): order_id = self._create_sales_order(self.strict_in_payload) response = self.client.get(f'/api/v1/sales-orders/{order_id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['id'], order_id) self.assertIn('human_id', response.data) order = business_models.SalesOrder.objects.get(id=order_id) expected_human_id = f"XS{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) def test_get_sales_order_detail_includes_order_quantity_on_items(self): payload = copy.deepcopy(self.strict_in_payload) payload['items'][0]['order_quantity'] = 88 order_id = self._create_sales_order(payload) response = self.client.get(f'/api/v1/sales-orders/{order_id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('items', response.data) self.assertEqual(len(response.data['items']), 1) self.assertIn('order_quantity', response.data['items'][0]) self.assertEqual(response.data['items'][0]['order_quantity'], 88) def test_update_sales_order_rejects_non_pending(self): order_id = self._create_sales_order(self.strict_in_payload) self.client.post( f'/api/v1/sales-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'order_date': '2025-12-02', 'items': [ { 'product_id': self.product.id, 'quantity': 10, 'num_of_rolls': 1, 'price': '30', } ], } response = self.client.put(f'/api/v1/sales-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不可修改', response.data['error']) def test_create_sales_order_with_printing_job_binding(self): payload = copy.deepcopy(self.relaxed_payload) payload['items'][0]['printing_job'] = self.printing_job_for_sales.id with patch('business.services.create_sales_order_stock_entries.delay'): response = self.client.post('/api/v1/sales-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) order = business_models.SalesOrder.objects.get(id=response.data['id']) item = order.items.first() self.assertEqual(item.printing_job_id, self.printing_job_for_sales.id) def test_review_sales_order_requires_action(self): order_id = self._create_sales_order(self.strict_in_payload) response = self.client.post(f'/api/v1/sales-orders/{order_id}/review/', {}, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('action', response.data) def test_review_sales_order_approve_success(self): order_id = self._create_sales_order(self.strict_in_payload) response = self.client.post( f'/api/v1/sales-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.SalesOrderStatusEnum.APPROVED) order = business_models.SalesOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.SalesOrderStatusEnum.APPROVED) def test_review_sales_order_cancel_blocked_after_stock_exists(self): order_id = self._create_sales_order(self.relaxed_payload) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.REMOVE, warehouse=self.warehouse_relaxed, source_type=stock_models.StockChangeSourceEnum.SALES, source_id=order_id, ) response = self.client.post( f'/api/v1/sales-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) order = business_models.SalesOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.SalesOrderStatusEnum.PENDING) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class PurchaseReturnOrderAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='采退商户', type=MerchantTypeEnum.FACTORY) self.supplier = Supplier.objects.create(merchant=self.merchant, name='供应商B') self.warehouse_strict = WareHouse.objects.create( merchant=self.merchant, name='采退严进仓', mode=WareHouseModeEnum.RESTRICT_IN, ) self.warehouse_relaxed = 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='RET', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='采退产品', human_id='RET-001', unit=ProductUnitEnum.METER, ) self.user = User.objects.create_user(username='return_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='仓管员', ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.strict_payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_strict.id, 'return_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'numbers': [5, 3], 'price': '11.5', 'unit': '米', } ], 'remarks': '采退测试', } self.strict_out_payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_strict_out.id, 'return_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'consume_detail_ids': [201, 202], 'quantity': 20, 'price': '11.2', 'unit': '米', } ], } def _create_purchase_return_order(self, payload): body = copy.deepcopy(payload) response = self.client.post('/api/v1/purchase-return-orders/', body, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) return response.data['id'] def test_create_purchase_return_order_success(self): response = self.client.post('/api/v1/purchase-return-orders/', self.strict_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) self.assertIn('human_id', response.data) self.assertEqual(response.data['status'], business_models.PurchaseReturnStatusEnum.PENDING) order = business_models.PurchaseReturnOrder.objects.get(id=response.data['id']) expected_human_id = f"CT{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) def test_create_purchase_return_requires_consume_ids_in_strict_out(self): payload = copy.deepcopy(self.strict_out_payload) payload['items'][0].pop('consume_detail_ids') response = self.client.post('/api/v1/purchase-return-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('consume_detail_ids', response.data['error']) def test_review_purchase_return_order_approve(self): order_id = self._create_purchase_return_order(self.strict_payload) response = self.client.post( f'/api/v1/purchase-return-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.PurchaseReturnStatusEnum.APPROVED) def test_purchase_return_cancel_blocked_after_stock_exists(self): order_id = self._create_purchase_return_order(self.strict_payload) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.REMOVE, warehouse=self.warehouse_strict, source_type=stock_models.StockChangeSourceEnum.PURCHASE_RETURN, source_id=order_id, ) response = self.client.post( f'/api/v1/purchase-return-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) order = business_models.PurchaseReturnOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.PurchaseReturnStatusEnum.PENDING) def test_update_purchase_return_order_success(self): order_id = self._create_purchase_return_order(self.strict_payload) payload = { 'supplier': self.supplier.id, 'warehouse': self.warehouse_relaxed.id, 'return_date': '2025-12-05', 'items': [ { 'product_id': self.product.id, 'quantity': 50, 'num_of_rolls': 2, 'price': '12.0', } ], 'remarks': '更新采退', } response = self.client.put(f'/api/v1/purchase-return-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['remarks'], '更新采退') def test_update_purchase_return_order_rejects_non_pending(self): order_id = self._create_purchase_return_order(self.strict_payload) self.client.post( f'/api/v1/purchase-return-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) payload = copy.deepcopy(self.strict_payload) response = self.client.put(f'/api/v1/purchase-return-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不可修改', response.data['error']) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class SalesReturnOrderAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='销退商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='销退客户', mobile='13900000000', created_by=None, ) self.warehouse_strict = WareHouse.objects.create( merchant=self.merchant, name='销退严进仓', mode=WareHouseModeEnum.RESTRICT_IN, ) self.warehouse_relaxed = WareHouse.objects.create( merchant=self.merchant, name='销退宽进仓', mode=WareHouseModeEnum.UNRESTRICTED, ) category = ProductCategory.objects.create( merchant=self.merchant, name='销退品类', product_prefix='SR', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='销退产品', human_id='SR-001', unit=ProductUnitEnum.METER, ) self.user = User.objects.create_user(username='sales_return_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='销退员', ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.strict_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_strict.id, 'return_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'numbers': [7, 3], 'price': '14.5', 'unit': '米', } ], } self.relaxed_payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'return_date': '2025-11-26', 'items': [ { 'product_id': self.product.id, 'quantity': 60, 'num_of_rolls': 2, 'price': '13.8', } ], 'remarks': '销退备注', } def _create_sales_return_order(self, payload): body = copy.deepcopy(payload) response = self.client.post('/api/v1/sales-return-orders/', body, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) return response.data['id'] def test_create_sales_return_order_success(self): response = self.client.post('/api/v1/sales-return-orders/', self.strict_payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('human_id', response.data) self.assertEqual(response.data['status'], business_models.SalesReturnStatusEnum.PENDING) order = business_models.SalesReturnOrder.objects.get(id=response.data['id']) expected_human_id = f"XT{order.created_at.strftime('%Y%m%d')}{order.id:06d}" self.assertEqual(response.data['human_id'], expected_human_id) def test_create_sales_return_order_invalid_customer(self): payload = {**self.strict_payload, 'customer': 999} response = self.client.post('/api/v1/sales-return-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_review_sales_return_order_approve(self): order_id = self._create_sales_return_order(self.relaxed_payload) response = self.client.post( f'/api/v1/sales-return-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.SalesReturnStatusEnum.APPROVED) def test_sales_return_cancel_blocked_after_stock_exists(self): order_id = self._create_sales_return_order(self.strict_payload) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_strict, source_type=stock_models.StockChangeSourceEnum.SALES_RETURN, source_id=order_id, ) response = self.client.post( f'/api/v1/sales-return-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) order = business_models.SalesReturnOrder.objects.get(id=order_id) self.assertEqual(order.status, business_models.SalesReturnStatusEnum.PENDING) def test_update_sales_return_order_success(self): order_id = self._create_sales_return_order(self.strict_payload) payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'return_date': '2025-12-02', 'items': [ { 'product_id': self.product.id, 'quantity': 40, 'num_of_rolls': 2, 'price': '18.0', } ], 'remarks': '更新销退', } response = self.client.put(f'/api/v1/sales-return-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['remarks'], '更新销退') def test_update_sales_return_order_rejects_non_pending(self): order_id = self._create_sales_return_order(self.strict_payload) self.client.post( f'/api/v1/sales-return-orders/{order_id}/review/', {'action': 'approve'}, format='json', ) payload = { 'customer': self.customer.id, 'warehouse': self.warehouse_relaxed.id, 'return_date': '2025-12-03', 'items': [ { 'product_id': self.product.id, 'quantity': 20, 'num_of_rolls': 1, 'price': '17', } ], } response = self.client.put(f'/api/v1/sales-return-orders/{order_id}/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('不可修改', response.data['error']) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class StockChangeOffsetAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='库存商户', type=MerchantTypeEnum.FACTORY) self.user = User.objects.create_user(username='stock_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='库存员', ) self.client = APIClient() self.client.force_authenticate(user=self.user) self.warehouse = WareHouse.objects.create( merchant=self.merchant, name='库存仓', mode=WareHouseModeEnum.RESTRICT_IN, ) category = ProductCategory.objects.create( merchant=self.merchant, name='库存品类', product_prefix='STK', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='库存产品', human_id='STK-001', unit=ProductUnitEnum.METER, ) self.stock_record, _, _ = stock_services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=stock_models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse.id, source_type=stock_models.StockChangeSourceEnum.PURCHASE, source_id=5001, products=[{'product': self.product.id, 'quantity': [Decimal('12')]}], ) stock_services.make_stock_change_completed(self.stock_record) def test_offset_stock_change_success(self): response = self.client.post( f'/api/v1/stock-change/{self.stock_record.id}/offset/', {'reason': '数据修正'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data['status'], 'success') self.assertEqual( stock_models.StockSnapshot.objects.filter( stock_change_record=self.stock_record, cancelled=True, ).count(), 1, ) def test_offset_stock_change_duplicate(self): self.client.post( f'/api/v1/stock-change/{self.stock_record.id}/offset/', {'reason': '首次'}, format='json', ) response = self.client.post( f'/api/v1/stock-change/{self.stock_record.id}/offset/', {'reason': '再次'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('已执行红冲', response.data['error']) class PaymentOrderAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='付款商户', type=MerchantTypeEnum.FACTORY) self.supplier = Supplier.objects.create(merchant=self.merchant, name='付款供应商') self.bank_account = BankAccount.objects.create( merchant=self.merchant, name='主账户', auto_number='BANK001', ) self.user = User.objects.create_user(username='pay_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 = { 'supplier': self.supplier.id, 'bank_account': self.bank_account.id, 'payment_date': '2025-11-26', 'amount': '120.5', 'discount_amount': '5.50', 'remarks': '付款备注', 'markup': '附言', } def test_create_payment_order_success(self): response = self.client.post('/api/v1/payment-orders/', self.payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) def test_create_payment_order_invalid_supplier(self): payload = {**self.payload, 'supplier': 999} response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_review_payment_order(self): resp = self.client.post('/api/v1/payment-orders/', self.payload, format='json') order_id = resp.data['id'] response = self.client.post(f'/api/v1/payment-orders/{order_id}/review/', {'action': 'approve'}, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.PaymentOrderStatusEnum.APPROVED) self.assertEqual(response.data['discount_amount'], '5.50') self.assertEqual(response.data['settlement_amount'], '126.00') response_cancel = self.client.post( f'/api/v1/payment-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response_cancel.status_code, status.HTTP_400_BAD_REQUEST) def test_payment_order_requires_amount(self): payload = {**self.payload} payload.pop('amount') response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('缺少 amount', response.data['error']) def test_payment_order_amount_must_be_positive(self): payload = {**self.payload, 'amount': '0'} response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('amount 不能为 0', response.data['error']) def test_payment_order_invalid_bank_account(self): other_merchant = Merchant.objects.create(name='其他商户', type=MerchantTypeEnum.FACTORY) other_bank = BankAccount.objects.create(merchant=other_merchant, name='其他账户', auto_number='BANK999') payload = {**self.payload, 'bank_account': other_bank.id} response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('银行账户不存在', response.data['error']) def test_payment_order_stores_bank_account_and_markup(self): response = self.client.post('/api/v1/payment-orders/', self.payload, format='json') order = business_models.PaymentOrder.objects.get(id=response.data['id']) self.assertEqual(order.bank_account_id, self.bank_account.id) self.assertEqual(order.markup, '附言') self.assertEqual(str(order.discount_amount), '5.50') self.assertEqual(str(order.settlement_amount), '126.00') def test_payment_order_discount_can_exceed_amount(self): payload = {**self.payload, 'discount_amount': '200.00'} response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) order = business_models.PaymentOrder.objects.get(id=response.data['id']) self.assertEqual(str(order.discount_amount), '200.00') self.assertEqual(str(order.settlement_amount), '320.50') def test_payment_order_discount_cannot_be_negative(self): payload = {**self.payload, 'discount_amount': '-1'} response = self.client.post('/api/v1/payment-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('discount_amount 不能小于 0', response.data['error']) class ReceiptOrderAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='收款商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='客户C', mobile='13812345678', created_by=None, ) self.bank_account = BankAccount.objects.create( merchant=self.merchant, name='收款账户', auto_number='BANK100', ) self.user = User.objects.create_user(username='receipt_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, 'bank_account': self.bank_account.id, 'receipt_date': '2025-11-26', 'amount': '88.00', 'discount_amount': '3.00', 'remarks': '收款备注', 'markup': '收款附言', } def test_create_receipt_order_success(self): response = self.client.post('/api/v1/receipt-orders/', self.payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertIn('id', response.data) def test_create_receipt_order_invalid_customer(self): payload = {**self.payload, 'customer': 999} response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_review_receipt_order(self): resp = self.client.post('/api/v1/receipt-orders/', self.payload, format='json') order_id = resp.data['id'] response = self.client.post( f'/api/v1/receipt-orders/{order_id}/review/', {'action': 'cancel'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['status'], business_models.ReceiptOrderStatusEnum.CANCELLED) self.assertEqual(response.data['discount_amount'], '3.00') self.assertEqual(response.data['settlement_amount'], '91.00') def test_receipt_order_requires_amount(self): payload = {**self.payload} payload.pop('amount') response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('缺少 amount', response.data['error']) def test_receipt_order_amount_must_be_positive(self): payload = {**self.payload, 'amount': '0'} response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('amount 不能为 0', response.data['error']) def test_receipt_order_stores_bank_account_and_markup(self): response = self.client.post('/api/v1/receipt-orders/', self.payload, format='json') order = business_models.ReceiptOrder.objects.get(id=response.data['id']) self.assertEqual(order.bank_account_id, self.bank_account.id) self.assertEqual(order.markup, '收款附言') self.assertEqual(str(order.discount_amount), '3.00') self.assertEqual(str(order.settlement_amount), '91.00') def test_receipt_order_discount_can_exceed_amount(self): payload = {**self.payload, 'discount_amount': '150.00'} response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) order = business_models.ReceiptOrder.objects.get(id=response.data['id']) self.assertEqual(str(order.settlement_amount), '238.00') def test_receipt_order_discount_cannot_be_negative(self): payload = {**self.payload, 'discount_amount': '-1'} response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('discount_amount 不能小于 0', response.data['error']) def test_receipt_order_invalid_bank_account(self): other_merchant = Merchant.objects.create(name='其他收款商户', type=MerchantTypeEnum.FACTORY) other_bank = BankAccount.objects.create(merchant=other_merchant, name='其他收款账户', auto_number='BANK888') payload = {**self.payload, 'bank_account': other_bank.id} response = self.client.post('/api/v1/receipt-orders/', payload, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('银行账户不存在', response.data['error']) class CustomerBalanceAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='余额商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='余额客户', mobile='13800000000', created_by=None, ) self.user = User.objects.create_user(username='balance_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) def test_customer_balance_defaults_to_zero(self): response = self.client.get(f'/api/v1/customers/{self.customer.id}/balance/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['balance'], '0') def test_customer_balance_reflects_sales_and_receipts(self): # 创建销售单 product_category = ProductCategory.objects.create( merchant=self.merchant, name='余额品类', product_prefix='BAL', ) product = Product.objects.create( merchant=self.merchant, category=product_category, name='余额产品', human_id='BAL-001', unit=ProductUnitEnum.METER, ) warehouse = WareHouse.objects.create( merchant=self.merchant, name='余额仓', mode=WareHouseModeEnum.RESTRICT_IN, ) order = services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date='2025-11-26', warehouse=warehouse, operator=self.employee, items=[{'product_id': product.id, 'numbers': [5], 'price': '10', 'unit': '米'}], ) services.review_sales_order( sales_order=order, target_status=business_models.SalesOrderStatusEnum.APPROVED, reviewed_by=self.user, ) response = self.client.get(f'/api/v1/customers/{self.customer.id}/balance/') self.assertEqual(response.data['balance'], '50.00') receipt = services.create_receipt_order( merchant=self.merchant, customer=self.customer, receipt_date='2025-11-27', amount='20', operator=self.employee, ) services.review_receipt_order( receipt_order=receipt, target_status=business_models.ReceiptOrderStatusEnum.APPROVED, reviewed_by=self.user, ) response = self.client.get(f'/api/v1/customers/{self.customer.id}/balance/') self.assertEqual(response.data['balance'], '30.00') class SupplierBalanceAPITestCase(TestCase): def setUp(self): self.merchant = Merchant.objects.create(name='供应商余额商户', type=MerchantTypeEnum.FACTORY) self.supplier = Supplier.objects.create(merchant=self.merchant, name='供应商余额') self.user = User.objects.create_user(username='supplier-balance', 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) def test_supplier_balance_api(self): response = self.client.get(f'/api/v1/suppliers/{self.supplier.id}/balance/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['supplier'], self.supplier.id) class StatementRecordAPITestCase(TestCase): """对账单单条记录查询 API 测试""" def setUp(self): self.merchant = Merchant.objects.create(name='对账商户', type=MerchantTypeEnum.FACTORY) self.customer = Customer.objects.create( merchant=self.merchant, name='对账客户', mobile='13800000000', created_by=None, ) self.supplier = Supplier.objects.create(merchant=self.merchant, name='对账供应商') category = ProductCategory.objects.create( merchant=self.merchant, name='对账品类', product_prefix='STM', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='对账产品', human_id='STM-001', unit=ProductUnitEnum.METER, ) self.warehouse = WareHouse.objects.create( merchant=self.merchant, name='对账仓库', mode=WareHouseModeEnum.UNRESTRICTED, ) self.user = User.objects.create_user(username='statement_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.sales_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date='2025-11-20', operator=self.employee, warehouse=self.warehouse, status=business_models.SalesOrderStatusEnum.APPROVED, ) business_models.SalesOrderItem.objects.create( sales_order=self.sales_order, product=self.product, price=Decimal('12.50'), color='红', quantity=Decimal('5'), unit='米', empty_diff_percent=Decimal('0'), quantity_of_rolls='2,3', num_of_rolls=2, spec='32S', ) self.purchase_order = business_models.PurchaseOrder.objects.create( merchant=self.merchant, supplier=self.supplier, purchase_date='2025-11-21', operator=self.employee, warehouse=self.warehouse, status=business_models.PurchaseOrderStatusEnum.APPROVED, ) business_models.PurchaseOrderItem.objects.create( purchase_order=self.purchase_order, product=self.product, price=Decimal('8.30'), color='蓝', quantity=Decimal('10'), unit='米', empty_diff_percent=Decimal('0'), quantity_of_rolls='6,4', num_of_rolls=2, spec='40S', ) def test_get_customer_sales_order_record(self): response = self.client.get( '/api/v1/statements/record/', { 'counterparty_type': 'customer', 'counterparty_id': self.customer.id, 'order_type': 'sales_order', 'order_id': self.sales_order.id, }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['counterparty'], self.customer.id) self.assertEqual(response.data['counterparty_name'], self.customer.name) self.assertEqual(len(response.data['records']), 1) record = response.data['records'][0] self.assertEqual(record['source_type'], 'sales_order') self.assertEqual(record['source_id'], self.sales_order.id) self.assertEqual(record['counterparty'], self.customer.id) self.assertIsNotNone(record.get('warehouse')) self.assertEqual(record['warehouse']['id'], self.warehouse.id) self.assertEqual(record['warehouse']['name'], self.warehouse.name) self.assertTrue(record['items']) self.assertEqual(record['items'][0]['color'], '红') self.assertEqual(record['items'][0]['spec'], '32S') self.assertEqual(record['items'][0]['quantity_of_rolls'], [2, 3]) self.assertEqual(record['items'][0]['num_of_rolls'], 2) def test_get_supplier_purchase_order_record(self): response = self.client.get( '/api/v1/statements/record/', { 'counterparty_type': 'supplier', 'counterparty_id': self.supplier.id, 'order_type': 'purchase_order', 'order_id': self.purchase_order.id, }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) record = response.data['records'][0] self.assertEqual(record['source_type'], 'purchase_order') self.assertEqual(record['source_id'], self.purchase_order.id) self.assertEqual(record['counterparty'], self.supplier.id) self.assertIsNotNone(record.get('warehouse')) self.assertEqual(record['warehouse']['id'], self.warehouse.id) self.assertEqual(record['warehouse']['name'], self.warehouse.name) self.assertTrue(record['items']) self.assertEqual(record['items'][0]['color'], '蓝') self.assertEqual(record['items'][0]['spec'], '40S') self.assertEqual(record['items'][0]['quantity_of_rolls'], [6, 4]) self.assertEqual(record['items'][0]['num_of_rolls'], 2) def test_statement_record_not_found(self): response = self.client.get( '/api/v1/statements/record/', { 'counterparty_type': 'customer', 'counterparty_id': self.customer.id, 'order_type': 'sales_order', 'order_id': 9999, }, ) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) self.assertIn('未找到匹配的对账记录', response.data['error']) def test_invalid_order_type_returns_400(self): response = self.client.get( '/api/v1/statements/record/', { 'counterparty_type': 'customer', 'counterparty_id': self.customer.id, 'order_type': 'unknown', 'order_id': self.sales_order.id, }, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('order_type', response.data) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class CeleryTasksTestCase(TestCase): """验证 Celery 任务的执行结果""" def setUp(self): self.merchant = Merchant.objects.create(name='Celery商户', type=MerchantTypeEnum.FACTORY) category = ProductCategory.objects.create( merchant=self.merchant, name='Celery品类', product_prefix='CLY', ) for idx in range(3): Product.objects.create( merchant=self.merchant, category=category, name=f'Celery产品{idx}', human_id=f'CLY-{idx:03d}', unit=ProductUnitEnum.METER, ) def test_ping_task_returns_payload(self): # 已弃用的演示任务，测试移除 self.skipTest('deprecated demo task') def test_merchant_product_count(self): # 已弃用的演示任务，测试移除 self.skipTest('deprecated demo task') def test_backup_database_creates_file(self): tmpdir = Path(tempfile.mkdtemp()) self.addCleanup(lambda: shutil.rmtree(tmpdir, ignore_errors=True)) with patch('api_v1.tasks._dump_database_to_sql') as mock_dump: def fake_dump(path: Path): path.write_text('-- dummy sql\n', encoding='utf-8') mock_dump.side_effect = fake_dump result = tasks.backup_database.delay(output_dir=str(tmpdir), filename_prefix='test-backup') payload = result.get(timeout=10) backup_path = Path(payload['backup_path']) self.assertTrue(backup_path.exists()) self.assertTrue(backup_path.is_file()) self.assertEqual(backup_path.parent.resolve(), tmpdir.resolve()) self.assertEqual(backup_path.suffix, '.sql') self.assertTrue(backup_path.read_text(encoding='utf-8').strip()) class PrintingJobWorkStateAPITestCase(TestCase): """印染任务 work_state 字段 API 覆盖""" def setUp(self): self.merchant = Merchant.objects.create( name='印染工厂', type=MerchantTypeEnum.FACTORY, ) self.user = User.objects.create_user(username='factory_user', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='印染员', ) perms = Permission.objects.filter( codename__in=['add_printingjob', 'view_printingjob', 'change_printingjob'] ) self.user.user_permissions.set(perms) self.client = APIClient() self.client.force_authenticate(user=self.user) category = ProductCategory.objects.create( merchant=self.merchant, name='面料', product_prefix='FAB', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='棉布', human_id='FAB-100', unit=ProductUnitEnum.METER, ) self.customer = Customer.objects.create( merchant=self.merchant, name='客户A', ) self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='纯棉', width='150cm', ) def test_create_job_with_work_state(self): payload = { 'printing_order': self.printing_order.id, 'product': self.product.id, 'quantity': 10, 'unit': '米', 'work_state': printing_models.PrintingJobWorkStateEnum.WAITING_FOR_DELIVERY, } resp = self.client.post('/api/v1/printing-jobs/', payload, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) self.assertEqual( resp.data['work_state'], printing_models.PrintingJobWorkStateEnum.WAITING_FOR_DELIVERY, ) job_id = resp.data['id'] detail = self.client.get(f'/api/v1/printing-jobs/{job_id}/') # 客户可见性过滤：customer 未绑定 created_by/visible_employees 时，任务对该员工不可见 => 404 self.assertEqual(detail.status_code, status.HTTP_404_NOT_FOUND) # 将客户显式授权给当前 employee 后应可见 => 200 self.customer.created_by = self.employee self.customer.save(update_fields=['created_by']) detail = self.client.get(f'/api/v1/printing-jobs/{job_id}/') self.assertEqual(detail.status_code, status.HTTP_200_OK) self.assertEqual(detail.data['work_state_display'], '待送货') def test_update_job_work_state(self): job = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=5, unit='米', ) url = f'/api/v1/printing-jobs/{job.id}/' patch_resp = self.client.patch( url, {'work_state': printing_models.PrintingJobWorkStateEnum.FINISHED}, format='json', ) # merchant 隔离：job.merchant 为空时，不属于当前员工商户 => 404 self.assertEqual(patch_resp.status_code, status.HTTP_404_NOT_FOUND) # 创建一个显式属于当前商户的 job，并让 customer 对 employee 可见后应可更新/可查询 self.customer.created_by = self.employee self.customer.save(update_fields=['created_by']) visible_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.printing_order, product=self.product, quantity=5, unit='米', ) visible_url = f'/api/v1/printing-jobs/{visible_job.id}/' patch_resp = self.client.patch( visible_url, {'work_state': printing_models.PrintingJobWorkStateEnum.FINISHED}, format='json', ) self.assertEqual(patch_resp.status_code, status.HTTP_200_OK) visible_job.refresh_from_db() self.assertEqual(visible_job.work_state, printing_models.PrintingJobWorkStateEnum.FINISHED) detail = self.client.get(visible_url) self.assertEqual(detail.status_code, status.HTTP_200_OK) self.assertEqual(detail.data['work_state_display'], '已完结') class PrintingJobBilledQuantityTestCase(TestCase): """验证印染任务的开单数量汇总""" def setUp(self): self.merchant = Merchant.objects.create( name='印染工厂B', type=MerchantTypeEnum.FACTORY, ) self.user = User.objects.create_user(username='factory_user_b', password='pass123') self.employee = Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='操作员B', ) category = ProductCategory.objects.create( merchant=self.merchant, name='面料B', product_prefix='FAB', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='棉布B', human_id='FAB-200', unit=ProductUnitEnum.METER, ) self.customer = Customer.objects.create( merchant=self.merchant, name='客户B', ) self.warehouse = WareHouse.objects.create( merchant=self.merchant, name='仓库B', mode=WareHouseModeEnum.UNRESTRICTED, ) self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉布', width='150cm', ) self.job = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=100, unit='米', ) self.sales_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date=datetime.date.today(), operator=self.employee, warehouse=self.warehouse, ) def test_billed_quantity_empty(self): self.assertEqual(self.job.billed_quantity, Decimal('0')) def test_billed_quantity_sum(self): business_models.SalesOrderItem.objects.create( sales_order=self.sales_order, product=self.product, price=Decimal('10'), quantity=Decimal('12.5'), unit='米', empty_diff_percent=Decimal('0'), num_of_rolls=1, printing_job=self.job, ) business_models.SalesOrderItem.objects.create( sales_order=self.sales_order, product=self.product, price=Decimal('11'), quantity=Decimal('7.5'), unit='米', empty_diff_percent=Decimal('0'), num_of_rolls=1, printing_job=self.job, ) self.assertEqual(self.job.billed_quantity, Decimal('20.0'))