"""Settlement Service 测试""" from datetime import date, datetime from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from basic_info import models as basic_models from printing import models as printing_models from settlement import services from stateflow import models as stateflow_models User = get_user_model() class SettlementServiceTestCase(TestCase): def setUp(self): self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username='test-user', password='pass123' ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name='测试员工' ) self.customer1 = basic_models.Customer.objects.create( merchant=self.merchant, name='客户A', created_by=self.employee ) self.customer2 = basic_models.Customer.objects.create( merchant=self.merchant, name='客户B', created_by=self.employee ) self.settlement_date = date(2026, 2, 8) class GetPlateOrderQuerySetTestCase(SettlementServiceTestCase): def test_filters_by_merchant(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='其他客户' ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, plate_type='首版', production_method='定位' ) printing_models.PlateOrder.objects.create( merchant=other_merchant, customer=other_customer, plate_type='首版', production_method='定位' ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) self.assertEqual(queryset.count(), 1) self.assertEqual(queryset.first().customer.name, '客户A') def test_filters_by_plate_type_not_null(self): printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, plate_type='首版', production_method='定位' ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, plate_type=None, production_method='定位' ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) self.assertEqual(queryset.count(), 1) self.assertIsNotNone(queryset.first().plate_type) def test_applies_customer_visibility_filter(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 ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer_visible, plate_type='首版', production_method='定位' ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer_invisible, plate_type='首版', production_method='定位' ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) self.assertEqual(queryset.count(), 1) self.assertEqual(queryset.first().customer.name, '可见客户') def test_superuser_sees_all_customers(self): self.user.is_superuser = True self.user.save() other_employee = basic_models.Employee.objects.create( sys_user=User.objects.create_user('other', 'pass123'), merchant=self.merchant, name='其他员工' ) customer_invisible = basic_models.Customer.objects.create( merchant=self.merchant, name='不可见客户', created_by=other_employee ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer_invisible, plate_type='首版', production_method='定位' ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) self.assertEqual(queryset.count(), 1) class GetMonthDateRangeTestCase(TestCase): def test_returns_month_start_and_settlement_date(self): settlement_date = date(2026, 2, 8) month_start, end_date = services._get_month_date_range(settlement_date) self.assertEqual(month_start, date(2026, 2, 1)) self.assertEqual(end_date, date(2026, 2, 8)) def test_handles_month_start(self): settlement_date = date(2026, 2, 1) month_start, end_date = services._get_month_date_range(settlement_date) self.assertEqual(month_start, date(2026, 2, 1)) self.assertEqual(end_date, date(2026, 2, 1)) def test_handles_month_end(self): settlement_date = date(2026, 2, 28) month_start, end_date = services._get_month_date_range(settlement_date) self.assertEqual(month_start, date(2026, 2, 1)) self.assertEqual(end_date, date(2026, 2, 28)) def test_handles_different_months(self): settlement_date = date(2026, 3, 15) month_start, end_date = services._get_month_date_range(settlement_date) self.assertEqual(month_start, date(2026, 3, 1)) self.assertEqual(end_date, date(2026, 3, 15)) class AggregatePlateOrdersByCustomerAndTypeTestCase(SettlementServiceTestCase): def test_aggregates_by_customer_and_type(self): today = self.settlement_date yesterday = date(2026, 2, 7) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, 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.customer1, plate_type='修改', production_method='定位', plate_date=timezone.make_aware(datetime.combine(yesterday, datetime.min.time())) ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) aggregated = services._aggregate_plate_orders_by_customer_and_type( queryset, self.settlement_date ) self.assertEqual(len(aggregated), 2) def test_calculates_today_count(self): today = self.settlement_date printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, 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.customer1, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(date(2026, 2, 7), datetime.min.time())) ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) aggregated = services._aggregate_plate_orders_by_customer_and_type( queryset, self.settlement_date ) today_item = [item for item in aggregated if item['type'] == '首版-定位'][0] self.assertEqual(today_item['today'], 1) def test_calculates_current_month_count(self): today = self.settlement_date month_start = date(2026, 2, 1) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, 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.customer1, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(month_start, datetime.min.time())) ) printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(date(2026, 1, 31), datetime.min.time())) ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) aggregated = services._aggregate_plate_orders_by_customer_and_type( queryset, self.settlement_date ) today_item = [item for item in aggregated if item['type'] == '首版-定位'][0] self.assertEqual(today_item['current_month'], 2) def test_filters_null_production_method(self): today = self.settlement_date printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, 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.customer1, plate_type='首版', production_method=None, plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) queryset = services._get_plate_order_queryset( self.merchant.id, self.user ) self.assertEqual(queryset.count(), 1) self.assertEqual(queryset.first().production_method, '定位') class FilterZeroDataTestCase(SettlementServiceTestCase): def test_filters_all_zero_data(self): data = [ {'type': '首版-定位', 'today': 0, 'current_month': 0}, {'type': '修改-定位', 'today': 0, 'current_month': 0} ] result = services._filter_zero_data(data) self.assertEqual(len(result), 0) def test_keeps_non_zero_today(self): data = [ {'type': '首版-定位', 'today': 1, 'current_month': 0}, {'type': '修改-定位', 'today': 0, 'current_month': 0} ] result = services._filter_zero_data(data) self.assertEqual(len(result), 1) self.assertEqual(result[0]['type'], '首版-定位') def test_keeps_non_zero_current_month(self): data = [ {'type': '首版-定位', 'today': 0, 'current_month': 5}, {'type': '修改-定位', 'today': 0, 'current_month': 0} ] result = services._filter_zero_data(data) self.assertEqual(len(result), 1) self.assertEqual(result[0]['type'], '首版-定位') class FormatPlateOrderSummaryTestCase(SettlementServiceTestCase): def test_formats_aggregated_data(self): aggregated = [ { 'customer__id': self.customer1.id, 'customer__name': '客户A', 'type': '首版-定位', 'today': 5, 'current_month': 10 }, { 'customer__id': self.customer1.id, 'customer__name': '客户A', 'type': '修改-定位', 'today': 2, 'current_month': 3 } ] result = services._format_plate_order_summary(aggregated) self.assertEqual(len(result), 1) self.assertEqual(result[0]['client_id'], self.customer1.id) self.assertEqual(result[0]['client_name'], '客户A') self.assertEqual(len(result[0]['plate_order_count']), 2) def test_filters_customers_with_zero_data(self): aggregated = [ { 'customer__id': self.customer1.id, 'customer__name': '客户A', 'type': '首版-定位', 'today': 0, 'current_month': 0 } ] result = services._format_plate_order_summary(aggregated) self.assertEqual(len(result), 0) def test_groups_by_customer(self): aggregated = [ { 'customer__id': self.customer1.id, 'customer__name': '客户A', 'type': '首版-定位', 'today': 5, 'current_month': 10 }, { 'customer__id': self.customer2.id, 'customer__name': '客户B', 'type': '首版-定位', 'today': 3, 'current_month': 6 } ] result = services._format_plate_order_summary(aggregated) self.assertEqual(len(result), 2) self.assertEqual(result[0]['client_id'], self.customer1.id) self.assertEqual(result[0]['client_name'], '客户A') self.assertEqual(result[1]['client_id'], self.customer2.id) self.assertEqual(result[1]['client_name'], '客户B') def test_groups_same_name_customers_by_customer_id(self): aggregated = [ { 'customer__id': self.customer1.id, 'customer__name': '同名客户', 'type': '首版-定位', 'today': 1, 'current_month': 2 }, { 'customer__id': self.customer2.id, 'customer__name': '同名客户', 'type': '修改-定位', 'today': 3, 'current_month': 4 } ] result = services._format_plate_order_summary(aggregated) self.assertEqual(len(result), 2) self.assertNotEqual(result[0]['client_id'], result[1]['client_id']) class GetPlateOrderSummaryByCustomerTestCase(SettlementServiceTestCase): def test_returns_summary_for_multiple_customers(self): today = self.settlement_date printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, 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.customer2, plate_type='首版', production_method='匹布', plate_date=timezone.make_aware(datetime.combine(today, datetime.min.time())) ) result = services.get_plate_order_summary_by_customer( self.merchant.id, self.settlement_date, self.user ) self.assertEqual(len(result), 2) self.assertIn('client_id', result[0]) def test_filters_zero_data(self): today = self.settlement_date printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=self.customer1, plate_type='首版', production_method='定位', plate_date=timezone.make_aware(datetime.combine(date(2026, 1, 1), datetime.min.time())) ) result = services.get_plate_order_summary_by_customer( self.merchant.id, self.settlement_date, self.user ) self.assertEqual(len(result), 0) def test_raises_for_invalid_merchant_id(self): with self.assertRaises(ValueError): services.get_plate_order_summary_by_customer( merchant_id=0, settlement_date=self.settlement_date, user=self.user ) def test_raises_for_invalid_settlement_date(self): with self.assertRaises(ValueError): services.get_plate_order_summary_by_customer( merchant_id=self.merchant.id, settlement_date='2026-02-08', user=self.user ) class GetPlateOrderSummaryByDesignerTestCase(SettlementServiceTestCase): def setUp(self): super().setUp() 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='调色完成') def _create_plate_order(self, *, customer=None, plate_date=None, plate_type='首版', production_method='定位'): plate_order = printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer or self.customer1, plate_type=plate_type, production_method=production_method, plate_date=timezone.make_aware(datetime.combine(plate_date or self.settlement_date, 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, cancelled=False): state_log = stateflow_models.StateFlowRecord.objects.create( business_object=plate_order.business_object, state=state or self.drawing_done, completed_by=self.user, is_cancelled=cancelled, ) return stateflow_models.StateLogParameterRecord.objects.create( state_log=state_log, parameters={'设计师名称': designer_name}, ) def test_returns_summary_for_drawing_done_designer(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A') result = services.get_plate_order_summary_by_designer( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(len(result), 1) self.assertEqual(result[0]['designer_name'], '设计师A') self.assertEqual(result[0]['plate_order_count'][0]['type'], '首版-定位') self.assertEqual(result[0]['plate_order_count'][0]['today'], 1) self.assertEqual(result[0]['plate_order_count'][0]['current_month'], 1) def test_defaults_to_drawing_done_only(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A', state=self.color_done) result = services.get_plate_order_summary_by_designer( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result, []) class GetDesignerWorkflowTaskSummaryTestCase(SettlementServiceTestCase): def setUp(self): super().setUp() 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='画图中') def _create_plate_order(self, *, customer=None, plate_type='首版', production_method='定位'): plate_order = printing_models.PlateOrder.objects.create( merchant=self.merchant, customer=customer or self.customer1, plate_type=plate_type, production_method=production_method, plate_date=timezone.make_aware(datetime.combine(self.settlement_date, 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 _create_state_log(self, plate_order, *, state=None, cancelled=False): return stateflow_models.StateFlowRecord.objects.create( business_object=plate_order.business_object, state=state or self.drawing_done, completed_by=self.user, is_cancelled=cancelled, ) def _add_params(self, state_log, params): return stateflow_models.StateLogParameterRecord.objects.create( state_log=state_log, parameters=params, ) def _add_designer_param(self, plate_order, *, designer_name, state=None, cancelled=False): state_log = self._create_state_log( plate_order, state=state or self.drawing_done, cancelled=cancelled, ) return self._add_params(state_log, { '设计师名称': designer_name, '完成时间': '2026-02-08 10:00:00', }) def test_sums_task_quantity_by_state_and_type(self): first = self._create_plate_order() second = self._create_plate_order() first_log = self._create_state_log(first, state=self.drawing_done) second_log = self._create_state_log(second, state=self.color_done) self._add_params(first_log, { '设计师名称': '左威', '完成数量': '3', '完成时间': '2026-02-08 10:00:00', }) self._add_params(second_log, { '设计师名称': '左威', '完成数量': '2', '完成时间': '2026-02-08 11:00:00', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, state_names=['画图完成', '调色完成'], ) self.assertEqual(len(result), 1) self.assertEqual(result[0]['designer_name'], '左威') self.assertEqual(result[0]['today'], 5) self.assertEqual(result[0]['current_month'], 5) by_state = {item['state_name']: item for item in result[0]['task_count']} self.assertEqual(by_state['画图完成']['today'], 3) self.assertEqual(by_state['调色完成']['today'], 2) def test_default_state_filter_includes_only_done_suffix(self): for state, quantity in [ (self.drawing_done, '3'), (self.color_done, '2'), (self.drawing_in_progress, '10'), ]: plate_order = self._create_plate_order() state_log = self._create_state_log(plate_order, state=state) self._add_params(state_log, { '设计师名称': '左威', '完成数量': quantity, '完成时间': '2026-02-08 10:00:00', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result[0]['today'], 5) states = {item['state_name'] for item in result[0]['task_count']} self.assertEqual(states, {'画图完成', '调色完成'}) def test_invalid_quantity_defaults_to_one_and_does_not_dedupe_plate_order(self): plate_order = self._create_plate_order() values = [None, '', '0', '-2', 'bad'] for index, value in enumerate(values): state_log = self._create_state_log(plate_order) self._add_params(state_log, { '设计师名称': '左威', '完成数量': value, '完成时间': f'2026-02-08 10:0{index}:00', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result[0]['today'], 5) self.assertEqual(result[0]['current_month'], 5) def test_multiple_parameter_records_for_one_state_log_count_once(self): plate_order = self._create_plate_order() state_log = self._create_state_log(plate_order) self._add_params(state_log, {'设计师名称': '左威'}) self._add_params(state_log, { '完成数量': '4', '完成时间': '2026-02-08 10:00:00', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result[0]['today'], 4) self.assertEqual(result[0]['current_month'], 4) def test_uses_completed_at_when_time_param_missing(self): plate_order = self._create_plate_order() state_log = self._create_state_log(plate_order) completed_at = timezone.make_aware(datetime(2026, 2, 8, 12, 0, 0)) stateflow_models.StateFlowRecord.objects.filter(id=state_log.id).update(completed_at=completed_at) state_log.refresh_from_db() self._add_params(state_log, { '设计师名称': '左威', '完成数量': '2', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result[0]['today'], 2) def test_filters_cancelled_and_designer_names(self): first = self._create_plate_order() second = self._create_plate_order() cancelled_log = self._create_state_log(first, cancelled=True) kept_log = self._create_state_log(second) self._add_params(cancelled_log, { '设计师名称': '左威', '完成数量': '10', '完成时间': '2026-02-08 10:00:00', }) self._add_params(kept_log, { '设计师名称': '王五', '完成数量': '3', '完成时间': '2026-02-08 10:00:00', }) result = services.get_designer_workflow_task_summary( self.merchant.id, self.settlement_date, self.user, designer_names=['王五'], ) self.assertEqual(len(result), 1) self.assertEqual(result[0]['designer_name'], '王五') self.assertEqual(result[0]['today'], 3) def test_accepts_custom_state_names(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A', state=self.color_done) result = services.get_plate_order_summary_by_designer( self.merchant.id, self.settlement_date, self.user, state_names=['画图完成', '调色完成'], ) self.assertEqual(len(result), 1) self.assertEqual(result[0]['designer_name'], '设计师A') def test_distincts_same_order_same_designer(self): plate_order = self._create_plate_order() self._add_designer_param(plate_order, designer_name='设计师A') self._add_designer_param(plate_order, designer_name='设计师A') result = services.get_plate_order_summary_by_designer( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result[0]['plate_order_count'][0]['today'], 1) self.assertEqual(result[0]['plate_order_count'][0]['current_month'], 1) def test_applies_customer_visibility(self): other_employee = basic_models.Employee.objects.create( sys_user=User.objects.create_user('designer-other', 'pass123'), merchant=self.merchant, name='其他员工' ) invisible_customer = basic_models.Customer.objects.create( merchant=self.merchant, name='不可见客户', created_by=other_employee, ) plate_order = self._create_plate_order(customer=invisible_customer) self._add_designer_param(plate_order, designer_name='设计师A') result = services.get_plate_order_summary_by_designer( self.merchant.id, self.settlement_date, self.user, ) self.assertEqual(result, [])