forked from erp-dev/erp
285 lines
11 KiB
Python
285 lines
11 KiB
Python
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 (
|
|
Employee,
|
|
Merchant,
|
|
MerchantTypeEnum,
|
|
Product,
|
|
ProductCategory,
|
|
ProductUnitEnum,
|
|
Supplier,
|
|
UserProfile,
|
|
WareHouse,
|
|
WareHouseModeEnum,
|
|
)
|
|
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 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)
|
|
mock_delay.assert_called_once()
|
|
|
|
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)
|
|
mock_delay.assert_called_once()
|
|
|
|
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('严进模式', response.data['error'])
|
|
|
|
|
|
@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())
|