"""Settlement API 测试""" from datetime import date, datetime from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from rest_framework import status from django.contrib.auth import get_user_model from basic_info import models as basic_models from printing import models as printing_models from stateflow import models as stateflow_models User = get_user_model() class PlateOrderSummaryAPITestCase(TestCase): """测试开版订单统计 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username='testuser', password='testpass123' ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name='测试员工' ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户', created_by=self.employee ) self.client.force_authenticate(user=self.user) def test_requires_date_parameter(self): """测试必须提供 date 参数""" response = self.client.get('/api/v1/settlement/plate-orders/summary/') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('缺少 date 参数', response.data['error']) def test_invalid_date_format(self): """测试无效的日期格式""" response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026/02/08' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('日期格式错误', response.data['error']) def test_invalid_date(self): """测试不存在的日期""" response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-30' ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('日期不存在', response.data['error']) def test_requires_merchant(self): """测试用户必须关联商户""" user_without_merchant = User.objects.create_user( username='no-merchant', password='pass123' ) self.client.force_authenticate(user=user_without_merchant) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('用户未关联商户', response.data['error']) def test_returns_summary_for_valid_date(self): """测试返回有效的统计数据""" today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('data', response.data) self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['client_id'], self.customer.id) self.assertEqual(response.data['data'][0]['client_name'], '测试客户') self.assertEqual(len(response.data['data'][0]['plate_order_count']), 1) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['type'], '首版-定位' ) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['today'], 1 ) def test_calculates_current_month(self): """测试计算本月累计数量""" today = date(2026, 2, 8) month_start = date(2026, 2, 1) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(month_start, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['current_month'], 2 ) def test_filters_null_plate_type(self): """测试过滤 plate_type 为空的订单""" today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type=None, production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['today'], 1 ) def test_filters_null_production_method(self): """测试过滤 production_method 为空的订单""" today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method=None, plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data['data']), 1) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['type'], '首版-定位' ) self.assertEqual( response.data['data'][0]['plate_order_count'][0]['today'], 1 ) def test_filters_zero_data(self): """测试过滤全0数据""" today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(date(2026, 1, 1), datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data['data']), 0) def test_applies_merchant_isolation(self): """测试应用商户隔离""" other_merchant = basic_models.Merchant.objects.create( name='其他商户', type=basic_models.MerchantTypeEnum.FACTORY ) other_customer = basic_models.Customer.objects.create( merchant=other_merchant, name='其他客户' ) today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=other_merchant, customer=other_customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['client_id'], self.customer.id) self.assertEqual(response.data['data'][0]['client_name'], '测试客户') def test_applies_customer_visibility(self): """测试应用客户可见性过滤""" other_employee = basic_models.Employee.objects.create( sys_user=User.objects.create_user('other', 'pass123'), merchant=self.merchant, name='其他员工' ) customer_visible = basic_models.Customer.objects.create( merchant=self.merchant, name='可见客户', created_by=other_employee ) customer_visible.visible_employees.set([self.employee]) customer_invisible = basic_models.Customer.objects.create( merchant=self.merchant, name='不可见客户', created_by=other_employee ) today = date(2026, 2, 8) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer_visible, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer_invisible, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['client_id'], customer_visible.id) self.assertEqual(response.data['data'][0]['client_name'], '可见客户') def test_requires_authentication(self): """测试需要认证""" self.client.force_authenticate(user=None) response = self.client.get( '/api/v1/settlement/plate-orders/summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) class PlateOrderDesignerSummaryAPITestCase(TestCase): """测试开版订单设计师统计 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username='designer-summary-user', password='testpass123' ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name='测试员工' ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户', created_by=self.employee ) self.process = stateflow_models.Process.objects.create(name='开版流程') self.drawing_done = stateflow_models.State.objects.create(name='画图完成') self.color_done = stateflow_models.State.objects.create(name='调色完成') self.client.force_authenticate(user=self.user) def _create_plate_order(self, *, plate_date=None): plate_order = printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(plate_date or date(2026, 2, 8), datetime.min.time())) ) business_object = stateflow_models.BusinessObject.objects.create( name=f'PlateOrder-{plate_order.id}', process=self.process, description='', ) plate_order.business_object = business_object plate_order.save(update_fields=['business_object']) return plate_order def _add_designer_param(self, plate_order, *, designer_name, state=None): state_log = stateflow_models.StateFlowRecord.objects.create( business_object=plate_order.business_object, state=state or self.drawing_done, completed_by=self.user, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=state_log, parameters={'设计师名称': designer_name}, ) def test_returns_designer_summary(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A') response = self.client.get( '/api/v1/settlement/plate-orders/designer-summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['meta']['state_names'], ['画图完成']) self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['designer_name'], '设计师A') self.assertEqual(response.data['data'][0]['plate_order_count'][0]['today'], 1) def test_accepts_state_names_query_param(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A', state=self.color_done) response = self.client.get( '/api/v1/settlement/plate-orders/designer-summary/?date=2026-02-08&state_names=调色完成' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['meta']['state_names'], ['调色完成']) self.assertEqual(response.data['data'][0]['designer_name'], '设计师A') def test_requires_date_parameter(self): response = self.client.get('/api/v1/settlement/plate-orders/designer-summary/') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('缺少 date 参数', response.data['error']) def test_invalid_state_names(self): response = self.client.get( '/api/v1/settlement/plate-orders/designer-summary/?date=2026-02-08&state_names=,,' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('state_names 不能为空', response.data['error']) class DesignerWorkflowTaskSummaryAPITestCase(TestCase): """测试设计师工序任务量统计 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username='designer-task-summary-user', password='testpass123' ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name='测试员工' ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户', created_by=self.employee ) self.process = stateflow_models.Process.objects.create(name='开版流程') self.drawing_done = stateflow_models.State.objects.create(name='画图完成') self.color_done = stateflow_models.State.objects.create(name='调色完成') self.drawing_in_progress = stateflow_models.State.objects.create(name='画图中') self.client.force_authenticate(user=self.user) def _create_plate_order(self): plate_order = printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(date(2026, 2, 8), datetime.min.time())) ) business_object = stateflow_models.BusinessObject.objects.create( name=f'PlateOrder-{plate_order.id}', process=self.process, description='', ) plate_order.business_object = business_object plate_order.save(update_fields=['business_object']) return plate_order def _add_task(self, *, designer_name='左威', quantity='3', state=None): plate_order = self._create_plate_order() state_log = stateflow_models.StateFlowRecord.objects.create( business_object=plate_order.business_object, state=state or self.drawing_done, completed_by=self.user, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=state_log, parameters={ '设计师名称': designer_name, '完成数量': quantity, '完成时间': '2026-02-08 10:00:00', }, ) def test_returns_workflow_task_summary(self): self._add_task(designer_name='左威', quantity='3') response = self.client.get( '/api/v1/settlement/workflows/designer-task-summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['meta']['state_names'], []) self.assertEqual(response.data['meta']['state_filter'], {'mode': 'suffix', 'suffix': '完成'}) self.assertEqual(response.data['meta']['quantity_param_key'], '完成数量') self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['designer_name'], '左威') self.assertEqual(response.data['data'][0]['today'], 3) self.assertEqual(response.data['data'][0]['task_count'][0]['state_name'], '画图完成') def test_default_state_filter_includes_only_done_suffix(self): self._add_task(designer_name='左威', quantity='3', state=self.drawing_done) self._add_task(designer_name='左威', quantity='2', state=self.color_done) self._add_task(designer_name='左威', quantity='10', state=self.drawing_in_progress) response = self.client.get( '/api/v1/settlement/workflows/designer-task-summary/?date=2026-02-08' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['data'][0]['today'], 5) states = {item['state_name'] for item in response.data['data'][0]['task_count']} self.assertEqual(states, {'画图完成', '调色完成'}) def test_accepts_state_and_designer_filters(self): self._add_task(designer_name='左威', quantity='3', state=self.drawing_done) self._add_task(designer_name='王五', quantity='2', state=self.color_done) response = self.client.get( '/api/v1/settlement/workflows/designer-task-summary/' '?date=2026-02-08&state_names=调色完成&designer_names=王五' ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['meta']['state_names'], ['调色完成']) self.assertEqual(response.data['meta']['state_filter'], {'mode': 'exact', 'state_names': ['调色完成']}) self.assertEqual(response.data['meta']['designer_names'], ['王五']) self.assertEqual(len(response.data['data']), 1) self.assertEqual(response.data['data'][0]['designer_name'], '王五') self.assertEqual(response.data['data'][0]['today'], 2) def test_requires_date_parameter(self): response = self.client.get('/api/v1/settlement/workflows/designer-task-summary/') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('缺少 date 参数', response.data['error']) def test_invalid_designer_names(self): response = self.client.get( '/api/v1/settlement/workflows/designer-task-summary/?date=2026-02-08&designer_names=,,' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('designer_names 不能为空', response.data['error'])