import copy import shutil import tempfile from pathlib import Path from unittest.mock import patch 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 ( Customer, Employee, EmployeeStatusEnum, Merchant, MerchantTypeEnum, Product, ProductCategory, ProductUnitEnum, Supplier, UserProfile, WareHouse, WareHouseModeEnum, ) from business import models as business_models, services from stock import models as stock_models from api_v1 import tasks 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.assertEqual(response.data['status'], 1) self.assertIn('等待审批', response.data['message']) mock_delay.assert_not_called() 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.assertEqual(response.data['status'], 1) 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) @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.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.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_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.assertEqual(response.data['status'], business_models.PurchaseReturnStatusEnum.PENDING) 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) @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.assertEqual(response.data['status'], business_models.SalesReturnStatusEnum.PENDING) 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) 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.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, 'payment_date': '2025-11-26', 'amount': '120.5', 'remarks': '付款备注', } 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) 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']) 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.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, 'receipt_date': '2025-11-26', 'amount': '88.00', 'remarks': '收款备注', } 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) 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']) 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') @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): result = tasks.ping_task.delay('celery hello') payload = result.get(timeout=5) self.assertEqual(payload['message'], 'celery hello') self.assertIn('timestamp', payload) self.assertIn('task_id', payload) def test_merchant_product_count(self): result = tasks.merchant_product_count.delay(self.merchant.id) payload = result.get(timeout=5) self.assertEqual(payload['merchant_id'], self.merchant.id) self.assertEqual(payload['product_count'], 3) 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())