from decimal import Decimal import datetime from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient, APIRequestFactory from basic_info import models as basic_models from printing import models as printing_models from business import models as business_models from api_v2.views.printing import PrintingJobByCustomerView from django.contrib.contenttypes.models import ContentType from stateflow import models as stateflow_models class QuickCreateEmployeeUserAPITest(TestCase): def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.admin_user = get_user_model().objects.create_user(username='admin', password='pass12345') self.admin_employee = basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.admin_user, name='管理员', ) self.client.force_authenticate(user=self.admin_user) self.url = '/api/v2/users/quick-create/' self.payload = { 'username': 'new_user', 'password': 'strongPass#1', 'display_name': '新员工', 'merchant_id': self.merchant.id, 'mobile': '13800000000', } def test_quick_create_employee_user_success(self): response = self.client.post(self.url, self.payload, format='json') self.assertEqual(response.status_code, 201) data = response.data self.assertIn('user', data) self.assertIn('employee', data) self.assertEqual(data['user']['username'], self.payload['username']) created_user = get_user_model().objects.get(username=self.payload['username']) self.assertEqual(created_user.employee.merchant, self.merchant) self.assertEqual(created_user.employee.name, self.payload['display_name']) def test_quick_create_employee_user_duplicate_username(self): get_user_model().objects.create_user(username=self.payload['username'], password='pass12345') response = self.client.post(self.url, self.payload, format='json') self.assertEqual(response.status_code, 400) self.assertIn('用户名已存在', str(response.data)) def test_quick_create_employee_user_invalid_merchant(self): payload = {**self.payload, 'merchant_id': 9999} response = self.client.post(self.url, payload, format='json') self.assertEqual(response.status_code, 400) self.assertIn('商户不存在', str(response.data)) class PrintingJobByCustomerAPITest(TestCase): def setUp(self): self.factory = APIRequestFactory() self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='印染商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户X', created_by=None, ) category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='品类', product_prefix='FAB', ) self.product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品A', human_id='FAB-001', width_size=Decimal('150.00'), color='红色', unit=basic_models.ProductUnitEnum.METER, ) self.other_product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品B', human_id='FAB-002', width_size=Decimal('160.00'), color='蓝色', unit=basic_models.ProductUnitEnum.METER, ) self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉', width='150cm', ) self.printing_order_other = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='麻', width='160cm', ) tz = timezone.get_default_timezone() in_range = timezone.make_aware(datetime.datetime(2025, 12, 5, 10, 0, 0), tz) out_range = timezone.make_aware(datetime.datetime(2025, 11, 20, 10, 0, 0), tz) self.job_in_range = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=10, unit='米', ) printing_models.PrintingJob.objects.filter(id=self.job_in_range.id).update(created_at=in_range) self.job_out_range = printing_models.PrintingJob.objects.create( printing_order=self.printing_order_other, product=self.other_product, quantity=20, unit='米', ) printing_models.PrintingJob.objects.filter(id=self.job_out_range.id).update(created_at=out_range) # 关联销售单，验证 billed_quantity warehouse = basic_models.WareHouse.objects.create( merchant=self.merchant, name='仓库A', mode=basic_models.WareHouseModeEnum.UNRESTRICTED, ) operator = basic_models.Employee.objects.create( merchant=self.merchant, name='操作员', ) sales_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date=datetime.date(2025, 12, 5), operator=operator, warehouse=warehouse, ) business_models.SalesOrderItem.objects.create( sales_order=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_in_range, ) business_models.SalesOrderItem.objects.create( sales_order=sales_order, product=self.product, price=Decimal('8'), quantity=Decimal('7.5'), unit='米', empty_diff_percent=Decimal('0'), num_of_rolls=1, printing_job=self.job_in_range, ) self.view = PrintingJobByCustomerView.as_view() def _get(self, params): request = self.factory.get('/api/v2/printing/jobs/', params) return self.view(request) def test_basic_date_and_customer_filter(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', }) self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['id'], self.job_in_range.id) self.assertEqual(resp.data[0]['billed_quantity'], '20.00') def test_filter_by_printing_order(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', 'printing_order': self.printing_order.id, }) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['printing_order'], self.printing_order.id) def test_filter_by_product_fields(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', 'product_id': self.product.id, 'product_name': '产品A', 'product_human_id': 'FAB-001', 'product_width_size': '150', 'product_color': '红', }) self.assertEqual(len(resp.data), 1) data = resp.data[0] self.assertEqual(data['product']['id'], self.product.id) self.assertEqual(data['billed_quantity'], '20.00') class PrintingJobBatchAdvanceV2APITest(TestCase): def setUp(self): self.client = APIClient() # 工厂用户（满足 IsPrintingFactory） self.merchant = basic_models.Merchant.objects.create( name='印染工厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='factory_user', password='pass12345') self.employee = basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='工厂员工', ) self.client.force_authenticate(user=self.user) # 基础客户/产品 self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户A', created_by=None, ) category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='品类', product_prefix='FAB', ) self.product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品A', human_id='FAB-100', width_size=Decimal('150.00'), color='红色', unit=basic_models.ProductUnitEnum.METER, ) # stateflow：流程 + 3个节点 + 节点参数（第一步必填） self.state1 = stateflow_models.State.objects.create(name='待处理1', description='第一个节点') self.state2 = stateflow_models.State.objects.create(name='待处理2', description='第二个节点') self.state3 = stateflow_models.State.objects.create(name='待处理3', description='第三个节点') self.param_required = stateflow_models.StateParameter.objects.create( key='temperature', value='', is_required=True, description='温度（必填）', ) self.state1.parameters.add(self.param_required) self.process = stateflow_models.Process.objects.create(name='印染流程') stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) # printing：订单 + 两条明细（每条明细都有 business_object） self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉', width='150cm', process=self.process, ) bo1 = stateflow_models.BusinessObject.objects.create( name='BO-1', process=self.process, ) bo2 = stateflow_models.BusinessObject.objects.create( name='BO-2', process=self.process, ) self.job1 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=10, unit='米', business_object=bo1, ) self.job2 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=20, unit='米', business_object=bo2, ) def test_preview_success_returns_next_state_and_parameters(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/preview/', {'printing_job_ids': [self.job1.id, self.job2.id]}, format='json' ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['printing_order_id'], self.printing_order.id) self.assertEqual(set(resp.data['printing_job_ids']), {self.job1.id, self.job2.id}) next_state = resp.data['next_state'] self.assertEqual(next_state['id'], self.state1.id) self.assertEqual(next_state['order'], 0) self.assertTrue(any(p['key'] == 'temperature' and p['is_required'] for p in next_state['parameters'])) def test_preview_fails_when_jobs_not_same_printing_order(self): other_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='麻', width='160cm', process=self.process, ) bo3 = stateflow_models.BusinessObject.objects.create(name='BO-3', process=self.process) job3 = printing_models.PrintingJob.objects.create( printing_order=other_order, product=self.product, quantity=5, unit='米', business_object=bo3, ) resp = self.client.post( '/api/v2/printing-jobs/batch-advance/preview/', {'printing_job_ids': [self.job1.id, job3.id]}, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('printing_order', resp.data.get('detail', '')) def test_submit_success_advances_all_and_creates_batch_record(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/', { 'printing_job_ids': [self.job1.id, self.job2.id], 'parameters': {'temperature': '25.5'} }, format='json' ) self.assertEqual(resp.status_code, 200) self.assertIn('batch_id', resp.data) # stateflow：两条 business_object 都应生成 1 条完成日志，且完成的是 state1 self.job1.refresh_from_db() self.job2.refresh_from_db() self.assertEqual(self.job1.business_object.state_logs.filter(is_cancelled=False).count(), 1) self.assertEqual(self.job2.business_object.state_logs.filter(is_cancelled=False).count(), 1) self.assertEqual(self.job1.business_object.state_logs.first().state_id, self.state1.id) self.assertEqual(self.job2.business_object.state_logs.first().state_id, self.state1.id) # printing：批量记录应存在且关联 jobs record = printing_models.PrintingJobBatchAdvanceRecord.objects.get(id=resp.data['batch_id']) self.assertEqual(record.created_by_id, self.user.id) self.assertEqual(record.printing_order_id, self.printing_order.id) self.assertEqual(record.state_id, self.state1.id) self.assertEqual(record.parameters.get('temperature'), '25.5') self.assertEqual(set(record.printing_jobs.values_list('id', flat=True)), {self.job1.id, self.job2.id}) def test_submit_fails_when_missing_required_parameter(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/', { 'printing_job_ids': [self.job1.id, self.job2.id], 'parameters': {} # 缺少 temperature }, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('缺失必填参数', resp.data.get('detail', '')) # 未产生任何日志/批量记录 self.assertEqual(self.job1.business_object.state_logs.count(), 0) self.assertEqual(self.job2.business_object.state_logs.count(), 0) self.assertEqual(printing_models.PrintingJobBatchAdvanceRecord.objects.count(), 0) class BusinessObjectCloneV2APITest(TestCase): def setUp(self): self.client = APIClient() self.user = get_user_model().objects.create_user(username='u1', password='pass12345') self.client.force_authenticate(user=self.user) self.state1 = stateflow_models.State.objects.create(name='S1') self.state2 = stateflow_models.State.objects.create(name='S2') self.process = stateflow_models.Process.objects.create(name='P1') stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) ct = ContentType.objects.get_for_model(stateflow_models.Process) self.bo = stateflow_models.BusinessObject.objects.create( name='BO', process=self.process, content_type=ct, object_id=111, ) # 创建一条日志 + 参数记录，确保“克隆包含工艺参数” log = stateflow_models.StateFlowRecord.objects.create( business_object=self.bo, state=self.state1, completed_by=self.user, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=log, parameters={'temperature': '25.5'}, remark='r1', ) def test_clone_business_object_api_returns_new_id(self): new_object_id = 222 resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': self.bo.content_type_id, 'object_id': new_object_id, }, format='json' ) self.assertEqual(resp.status_code, 200) self.assertIn('business_object_id', resp.data) new_id = resp.data['business_object_id'] self.assertNotEqual(new_id, self.bo.id) cloned = stateflow_models.BusinessObject.objects.get(id=new_id) self.assertEqual(cloned.process_id, self.bo.process_id) self.assertEqual(cloned.state_logs.count(), self.bo.state_logs.count()) self.assertEqual(cloned.content_type_id, self.bo.content_type_id) self.assertEqual(cloned.object_id, new_object_id) def test_clone_business_object_api_404(self): resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': 999999, 'content_type': self.bo.content_type_id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 404) def test_clone_business_object_api_rejects_mismatched_content_type(self): # 用另一个 ContentType（比如 State） ct_state = ContentType.objects.get_for_model(stateflow_models.State) self.assertNotEqual(ct_state.id, self.bo.content_type_id) resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': ct_state.id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('content_type', resp.data.get('detail', '')) def test_clone_business_object_api_unauthorized(self): client = APIClient() resp = client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': self.bo.content_type_id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 401)