from datetime import date from io import BytesIO from types import SimpleNamespace from openpyxl import load_workbook from app.routers import dashboard from app.services.excel_export import export_dashboard_rows def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc(): report_date = date(2026, 7, 25) report = SimpleNamespace( id=1, attendance_point_name="总厂", report_date=report_date, employee_phone="13800000000", employee=SimpleNamespace(name="张三", is_temporary=False), items=[], audit_logs=[], session=None, is_system_auto_submitted=False, is_voided=False, is_multi_person_assistant=False, review_remark="复核通过", ) item = SimpleNamespace( id=10, attendance_point_name="总厂", device_no="A-01", project_no="P1", product_name="23#", process_name="冲压", remark=None, raw_material_batch_no="BATCH-1", material_code="M-1", material_name="铝材", profile_no="PF-1", stamping_method="单冲", operator_count=1, process_unit_price_yuan=2, standard_beat=1, good_qty=650, defect_qty=0, scrap_qty=0, changeover_count=2, allocated_minutes=650, ) report.items = [item] allocations = [ SimpleNamespace( report=report, item=item, allocation_date=date(2026, 7, 24), night_minutes=350, overtime_minutes=0, day_minutes=0, effective_minutes=350, good_qty=350, defect_qty=0, scrap_qty=0, changeover_count=1, reference_wage=700, ), SimpleNamespace( report=report, item=item, allocation_date=date(2026, 7, 25), night_minutes=0, overtime_minutes=60, day_minutes=240, effective_minutes=300, good_qty=300, defect_qty=0, scrap_qty=0, changeover_count=1, reference_wage=600, ), ] rows = dashboard._dashboard_rows_from_allocations(allocations) assert [row.report_date for row in rows] == [date(2026, 7, 25), date(2026, 7, 24)] assert [row.source_report_dates for row in rows] == [[report_date], [report_date]] assert [row.source_report_date for row in rows] == [report_date, report_date] assert [row.report_count for row in rows] == [1, 1] current, previous = rows assert current.effective_minutes == 300 assert current.shift_day_minutes == 240 assert current.shift_overtime_minutes == 60 assert current.shift_night_minutes == 0 assert current.total_good_qty == 300 assert current.total_defect_qty == 0 assert current.total_output_qty == 300 assert current.changeover_count == 1 assert current.reference_wage == 600 assert previous.effective_minutes == 350 assert previous.shift_day_minutes == 0 assert previous.shift_overtime_minutes == 0 assert previous.shift_night_minutes == 350 assert previous.total_good_qty == 350 assert previous.total_defect_qty == 0 assert previous.total_output_qty == 350 assert previous.changeover_count == 1 assert previous.reference_wage == 700 def test_export_dashboard_rows_writes_allocation_and_source_dates_with_warning_columns(): row = SimpleNamespace( attendance_point_name="总厂", report_date=date(2026, 7, 24), source_report_dates=[date(2026, 7, 25), date(2026, 7, 23)], source_report_date=None, employee_name="张三", employee_phone="13800000000", worker_type="正式工", project_no="P1", product_name="23#", process_name="冲压", effective_minutes=60, shift_distribution_text="白班1小时", total_good_qty=50, total_defect_qty=2, expected_workload=100, actual_beat=20, standard_beat=10, profile_no="PF-1", material_code="M-1", material_name="铝材", raw_material_batch_no="BATCH-1", device_no="A-01", stamping_method="单冲", changeover_count=1, operator_count=1, process_unit_price_yuan=2, reference_wage=100, is_voided=False, is_multi_person=False, is_continuous_die=False, correction_pairs=[], review_remark="复核通过", ) workbook = load_workbook(BytesIO(export_dashboard_rows([row]))) sheet = workbook["报工看板"] assert [sheet.cell(row=1, column=index).value for index in range(1, 5)] == [ "考勤点", "统计归属日期", "原始报工日期", "员工姓名", ] assert sheet.cell(row=2, column=2).value == "2026-07-24" assert sheet.cell(row=2, column=3).value == "2026-07-23、2026-07-25" assert sheet.cell(row=2, column=12).fill.fgColor.rgb == "FFFF6666" assert sheet.cell(row=2, column=15).fill.fgColor.rgb == "FFFF6666" def test_dashboard_rows_include_over_limit_summary_from_reports(): report = SimpleNamespace( id=1, attendance_point_name="总厂", employee_phone="13800000000", report_date=date(2026, 7, 25), has_over_limit=True, over_limit_summary="1项超额", review_remark=None, employee=SimpleNamespace(name="张三", is_temporary=False), items=[], audit_logs=[], session=None, is_system_auto_submitted=False, is_voided=False, is_multi_person_assistant=False, ) item = SimpleNamespace( id=1, report=report, report_id=1, attendance_point_name="总厂", device_no="A", project_no="P1", product_name="产品A", material_code="M1", material_name="材料A", raw_material_batch_no="LOT-1", process_name="1", remark=None, stamping_method="冲压", operator_count=1, process_unit_price_yuan=1, standard_beat=1, good_qty=60, defect_qty=0, scrap_qty=0, changeover_count=0, allocated_minutes=60, over_limit_status="exceeded", over_limit_reason="1序累计报工数量超过本批次材料可支撑数量", ) report.items = [item] allocation = SimpleNamespace( report=report, item=item, report_id=1, allocation_date=date(2026, 7, 25), attendance_point_name="总厂", employee_phone="13800000000", good_qty=60, defect_qty=0, scrap_qty=0, effective_minutes=60, day_minutes=60, overtime_minutes=0, night_minutes=0, reference_wage=60, changeover_count=0, ) rows = dashboard._dashboard_rows_from_allocations([allocation]) assert rows[0].has_over_limit is True assert rows[0].over_limit_summary == "1项超额" assert rows[0].over_limit_reasons == "1序累计报工数量超过本批次材料可支撑数量" def test_export_dashboard_rows_writes_over_limit_columns(): row = SimpleNamespace( attendance_point_name="总厂", report_date=date(2026, 7, 24), source_report_dates=[date(2026, 7, 24)], source_report_date=None, employee_name="张三", employee_phone="13800000000", worker_type="正式工", project_no="P1", product_name="23#", process_name="1", effective_minutes=60, shift_distribution_text="白班1小时", total_good_qty=50, total_defect_qty=2, expected_workload=100, actual_beat=20, standard_beat=10, profile_no="PF-1", material_code="M-1", material_name="铝材", raw_material_batch_no="BATCH-1", device_no="A-01", stamping_method="单冲", changeover_count=1, operator_count=1, process_unit_price_yuan=2, reference_wage=100, has_over_limit=True, over_limit_summary="1项超额", over_limit_reasons="1序累计报工数量超过本批次材料可支撑数量", is_voided=False, is_multi_person=False, is_continuous_die=False, correction_pairs=[], review_remark="复核通过", ) workbook = load_workbook(BytesIO(export_dashboard_rows([row]))) sheet = workbook["报工看板"] headers = [sheet.cell(row=1, column=index).value for index in range(1, sheet.max_column + 1)] assert "是否超额" in headers assert "超额摘要" in headers assert "超额原因" in headers assert sheet.cell(row=2, column=headers.index("是否超额") + 1).value == "是" assert sheet.cell(row=2, column=headers.index("超额摘要") + 1).value == "1项超额" assert sheet.cell(row=2, column=headers.index("超额原因") + 1).value == "1序累计报工数量超过本批次材料可支撑数量" def test_dashboard_over_limit_does_not_spread_to_other_items_in_same_report(): report = SimpleNamespace( id=1, attendance_point_name="总厂", employee_phone="13800000000", report_date=date(2026, 7, 25), has_over_limit=True, over_limit_summary="1项超额", review_remark=None, employee=SimpleNamespace(name="张三", is_temporary=False), items=[], audit_logs=[], session=None, is_system_auto_submitted=False, is_voided=False, is_multi_person_assistant=False, ) exceeded_item = SimpleNamespace( id=1, report=report, report_id=1, attendance_point_name="总厂", device_no="A", project_no="P1", product_name="产品A", material_code="M1", material_name="材料A", raw_material_batch_no="LOT-1", process_name="1", remark=None, stamping_method="冲压", operator_count=1, process_unit_price_yuan=1, standard_beat=1, good_qty=60, defect_qty=0, scrap_qty=0, changeover_count=0, allocated_minutes=60, over_limit_status="exceeded", over_limit_reason="1序累计报工数量超过本批次材料可支撑数量", ) normal_item = SimpleNamespace( id=2, report=report, report_id=1, attendance_point_name="总厂", device_no="B", project_no="P1", product_name="产品A", material_code="M1", material_name="材料A", raw_material_batch_no="LOT-1", process_name="2", remark=None, stamping_method="冲压", operator_count=1, process_unit_price_yuan=1, standard_beat=1, good_qty=40, defect_qty=0, scrap_qty=0, changeover_count=0, allocated_minutes=40, over_limit_status="ok", over_limit_reason=None, ) report.items = [exceeded_item, normal_item] allocations = [ SimpleNamespace( report=report, item=exceeded_item, report_id=1, allocation_date=date(2026, 7, 25), attendance_point_name="总厂", employee_phone="13800000000", good_qty=60, defect_qty=0, scrap_qty=0, effective_minutes=60, day_minutes=60, overtime_minutes=0, night_minutes=0, reference_wage=60, changeover_count=0, ), SimpleNamespace( report=report, item=normal_item, report_id=1, allocation_date=date(2026, 7, 25), attendance_point_name="总厂", employee_phone="13800000000", good_qty=40, defect_qty=0, scrap_qty=0, effective_minutes=40, day_minutes=40, overtime_minutes=0, night_minutes=0, reference_wage=40, changeover_count=0, ), ] rows = dashboard._dashboard_rows_from_allocations(allocations) rows_by_process = {row.process_name: row for row in rows} assert rows_by_process["1"].has_over_limit is True assert rows_by_process["1"].over_limit_reasons == "1序累计报工数量超过本批次材料可支撑数量" assert rows_by_process["2"].has_over_limit is False assert rows_by_process["2"].over_limit_summary is None assert rows_by_process["2"].over_limit_reasons is None