forked from erp-dev/erp
1636 lines
68 KiB
Python
1636 lines
68 KiB
Python
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,
|
|
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_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)
|
|
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': '更新采购单',
|
|
}
|
|
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'], '更新采购单')
|
|
|
|
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_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}/')
|
|
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',
|
|
)
|
|
self.assertEqual(patch_resp.status_code, status.HTTP_200_OK)
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.work_state, printing_models.PrintingJobWorkStateEnum.FINISHED)
|
|
detail = self.client.get(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'))
|