erpnew/api_v1/tests.py

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 = WareHouse.objects.create(
            merchant=self.merchant,
            name='主仓',
            mode=WareHouseModeEnum.RESTRICT_IN,
        )
        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.payload = {
            'supplier': self.supplier.id,
            'warehouse': self.warehouse.id,
            'order_date': '2025-11-26',
            'total_amount': '1500.00',
            'items': [
                {
                    'product_id': self.product.id,
                    'quantities': ['10.0', '5.0'],
                }
            ],
            'remarks': '接口测试',
        }

    def test_create_purchase_order_success(self):
        with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay:
            response = self.client.post('/api/v1/purchase-orders/', self.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.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.payload, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)


@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())