from datetime import date, datetime from decimal import Decimal from types import SimpleNamespace import pytest from sqlalchemy import BigInteger, create_engine, inspect, select from sqlalchemy.ext.compiler import compiles from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import raiseload, selectinload, sessionmaker from app.database import Base from app.models import ( AttendancePoint, Equipment, Personnel, Product, ProductionReport, ProductionReportAllocation, ProductionReportItem, ReportStatus, Role, SessionStatus, WorkSession, WorkSessionDevice, ) from app.routers import reports as reports_router from app.routers import reviews as reviews_router from app.schemas import ( ApproveReportRequest, CleaningReportSubmitItem, CleaningReportSubmitRequest, ReportItemCorrection, ReportSubmitBlock, ReportSubmitItem, ReportSubmitRequest, ) from app.services import auto_submit as auto_submit_service from app.services import reporting_window as reporting_window_service from app.services.report_lifecycle import purge_expired_voided_reports from app.services.metrics import calculate_report_metrics from app.services.report_allocations import ( allocation_summary_text, build_report_allocation_drafts, refresh_report_allocations, ) from app.services.serializers import report_out from app.services.work_schedule import WorkScheduleConfig @compiles(BigInteger, "sqlite") def _compile_big_integer_for_sqlite(type_, compiler, **kw): return "INTEGER" def _item( *, id: int = 1, good_qty: float = 0, defect_qty: float = 0, scrap_qty: float = 0, changeover_count: float = 0, process_unit_price_yuan: float = 0, started_at: datetime | None = None, device_no: str = "23#", product_name: str | None = None, process_name: str | None = "冲压", ): return SimpleNamespace( id=id, device_no=device_no, product_name=product_name or device_no, process_name=process_name, started_at=started_at, standard_beat=1, standard_workload=0, stamping_method=None, good_qty=good_qty, defect_qty=defect_qty, scrap_qty=scrap_qty, changeover_count=changeover_count, process_unit_price_yuan=process_unit_price_yuan, allocated_minutes=0, ) def _device(scanned_at: datetime, *, device_no: str = "23#", process_name: str = "冲压", sort_order: int = 0): return SimpleNamespace( device_no=device_no, process_name=process_name, scanned_at=scanned_at, released_at=None, release_reason=None, sort_order=sort_order, ) def _report( start_at: datetime, end_at: datetime, item, *, id: int = 1, report_date: date | None = None, devices: list | None = None, ): return SimpleNamespace( id=id, attendance_point_name="总厂", employee_phone="13800000000", report_date=report_date or start_at.date(), start_at=start_at, end_at=end_at, items=item if isinstance(item, list) else [item], session=SimpleNamespace(devices=devices or []), ) def test_cross_day_after_midnight_allocates_previous_night_and_current_overtime_day(): item = _item( good_qty=650, defect_qty=65, scrap_qty=6.5, changeover_count=13, process_unit_price_yuan=1, ) report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) rows = build_report_allocation_drafts(report) assert len(rows) == 2 previous, current = rows assert previous.allocation_date == date(2026, 7, 24) assert previous.night_minutes == 350 assert previous.day_minutes == 0 assert previous.overtime_minutes == 0 assert previous.effective_minutes == 350 assert previous.good_qty == 350 assert previous.defect_qty == 35 assert previous.scrap_qty == 3.5 assert previous.changeover_count == 7 assert previous.reference_wage == 350 assert current.allocation_date == date(2026, 7, 25) assert current.night_minutes == 0 assert current.overtime_minutes == 120 assert current.day_minutes == 180 assert current.effective_minutes == 300 assert current.good_qty == 300 assert current.defect_qty == 30 assert current.scrap_qty == 3 assert current.changeover_count == 6 assert current.reference_wage == 300 def test_one_minute_before_night_end_allocates_to_previous_date(): item = _item(good_qty=301, process_unit_price_yuan=2) report = _report(datetime(2026, 7, 25, 5, 59), datetime(2026, 7, 25, 11, 0), item) rows = build_report_allocation_drafts(report) assert [(row.allocation_date, row.night_minutes, row.overtime_minutes, row.day_minutes) for row in rows] == [ (date(2026, 7, 24), 1, 0, 0), (date(2026, 7, 25), 0, 120, 180), ] assert rows[0].good_qty == 1 assert rows[0].reference_wage == 2 assert rows[1].good_qty == 300 assert rows[1].reference_wage == 600 def test_no_device_shared_full_report_splits_effective_minutes_between_items(): first_item = _item(id=1, good_qty=10) second_item = _item(id=2, good_qty=20) report = _report( datetime(2026, 7, 25, 8, 0), datetime(2026, 7, 25, 11, 0), [first_item, second_item], ) rows = build_report_allocation_drafts(report) assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [ (1, 90, 90), (2, 90, 90), ] def test_short_shared_report_rounding_residual_keeps_aggregate_night_total(): start_at = datetime(2026, 5, 3, 1, 4, 50) end_at = datetime(2026, 5, 3, 1, 4, 53) first_item = _item(id=1, good_qty=1) second_item = _item(id=2, good_qty=1) report = _report(start_at, end_at, [first_item, second_item]) metrics = calculate_report_metrics(start_at, end_at, [first_item, second_item]) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 0.05 assert metrics["shift_night_minutes"] == 0.05 assert [row.allocation_date for row in rows] == [date(2026, 5, 2), date(2026, 5, 2)] assert [row.effective_minutes for row in rows] == [0.03, 0.02] assert round(sum(row.effective_minutes for row in rows), 2) == 0.05 assert round(sum(row.night_minutes for row in rows), 2) == 0.05 assert round(sum(row.day_minutes for row in rows), 2) == 0 assert round(sum(row.overtime_minutes for row in rows), 2) == 0 assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows) def test_item_split_fields_keep_original_totals_after_rounding_residual(): item = _item( good_qty=1, defect_qty=1, scrap_qty=1, changeover_count=1, process_unit_price_yuan=1, ) report = _report(datetime(2026, 7, 25, 6, 0), datetime(2026, 7, 28, 6, 0), item) rows = build_report_allocation_drafts(report) assert [row.allocation_date for row in rows] == [ date(2026, 7, 25), date(2026, 7, 26), date(2026, 7, 27), ] assert [row.effective_minutes for row in rows] == [1340, 1340, 1340] assert round(sum(row.good_qty for row in rows), 2) == 1 assert round(sum(row.defect_qty for row in rows), 2) == 1 assert round(sum(row.scrap_qty for row in rows), 2) == 1 assert round(sum(row.changeover_count for row in rows), 2) == 1 assert round(sum(row.reference_wage for row in rows), 2) == 1 def test_reference_wage_uses_half_up_rounding_for_allocation_totals(): item = _item(good_qty=115, process_unit_price_yuan=Decimal("0.867")) report = _report(datetime(2026, 7, 25, 8, 0), datetime(2026, 7, 25, 10, 0), item) rows = build_report_allocation_drafts(report) assert len(rows) == 1 assert rows[0].reference_wage == 99.71 def test_split_reference_wage_residual_keeps_half_up_item_total(): item = _item(good_qty=115, process_unit_price_yuan=Decimal("0.867")) report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) rows = build_report_allocation_drafts(report) assert [row.allocation_date for row in rows] == [date(2026, 7, 24), date(2026, 7, 25)] assert round(sum(row.reference_wage for row in rows), 2) == 99.71 def test_items_sharing_same_started_at_split_the_segment_minutes(): started_at = datetime(2026, 7, 25, 8, 0) first_item = _item(id=1, good_qty=10, started_at=started_at) second_item = _item(id=2, good_qty=20, started_at=started_at) report = _report( started_at, datetime(2026, 7, 25, 11, 0), [first_item, second_item], devices=[_device(started_at)], ) rows = build_report_allocation_drafts(report) assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [ (1, 90, 90), (2, 90, 90), ] def test_device_matched_and_unmatched_items_match_metrics_remaining_minutes(): start_at = datetime(2026, 7, 25, 8, 0) end_at = datetime(2026, 7, 25, 11, 0) matching_item = _item(id=1, good_qty=10, device_no="23#") unmatched_item = _item(id=2, good_qty=20, device_no="99#") report = _report(start_at, end_at, [matching_item, unmatched_item], devices=[_device(start_at)]) calculate_report_metrics(start_at, end_at, [matching_item, unmatched_item], devices=[_device(start_at)]) rows = build_report_allocation_drafts(report) assert {matching_item.id: matching_item.allocated_minutes, unmatched_item.id: unmatched_item.allocated_minutes} == { 1: 180, 2: 0, } assert {row.report_item_id: row.effective_minutes for row in rows} == { 1: 180, 2: 0, } def test_unassigned_device_segment_is_allocated_when_all_items_already_matched(): start_at = datetime(2026, 7, 25, 0, 0) switch_at = datetime(2026, 7, 25, 8, 0) end_at = datetime(2026, 7, 25, 10, 0) devices = [ _device(start_at, device_no="23#", sort_order=0), _device(switch_at, device_no="99#", sort_order=1), ] item = _item(id=1, good_qty=600, device_no="23#", started_at=start_at) report = _report(start_at, end_at, item, devices=devices) metrics = calculate_report_metrics(start_at, end_at, [item], devices=devices) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 600 assert metrics["shift_day_minutes"] == 120 assert metrics["shift_overtime_minutes"] == 120 assert metrics["shift_night_minutes"] == 360 assert round(sum(row.effective_minutes for row in rows), 2) == metrics["effective_minutes"] assert { "day": round(sum(row.day_minutes for row in rows), 2), "overtime": round(sum(row.overtime_minutes for row in rows), 2), "night": round(sum(row.night_minutes for row in rows), 2), } == { "day": metrics["shift_day_minutes"], "overtime": metrics["shift_overtime_minutes"], "night": metrics["shift_night_minutes"], } assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows) def test_night_allocation_uses_meal_deducted_minutes_with_previous_date_rules(): schedule = WorkScheduleConfig( day_start="08:00", day_end="17:20", lunch_start="11:40", lunch_end="12:40", dinner_start="21:00", dinner_end="21:30", overtime_start="18:00", overtime_end="20:00", night_start="20:00", night_end="06:00", ) start_at = datetime(2026, 7, 25, 20, 0) end_at = datetime(2026, 7, 25, 22, 0) item = _item(id=1, good_qty=90) report = _report(start_at, end_at, item) metrics = calculate_report_metrics(start_at, end_at, [item], schedule=schedule) rows = build_report_allocation_drafts(report, schedule=schedule) assert metrics["effective_minutes"] == 90 assert metrics["shift_night_minutes"] == 90 assert [(row.allocation_date, row.night_minutes, row.effective_minutes) for row in rows] == [ (date(2026, 7, 25), 90, 90), ] def test_blank_overtime_meal_deduction_is_clipped_like_metrics(): start_at = datetime(2026, 7, 25, 6, 0) end_at = datetime(2026, 7, 25, 19, 0) item = _item(id=1, good_qty=680) report = _report(start_at, end_at, item) metrics = calculate_report_metrics(start_at, end_at, [item]) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 680 assert metrics["shift_day_minutes"] == 500 assert metrics["shift_overtime_minutes"] == 180 assert [(row.effective_minutes, row.day_minutes, row.overtime_minutes, row.night_minutes) for row in rows] == [ (680, 500, 180, 0), ] def test_same_device_item_split_inside_meal_is_rescaled_to_metrics_effective_minutes(): start_at = datetime(2026, 7, 25, 8, 0) second_started_at = datetime(2026, 7, 25, 12, 0) end_at = datetime(2026, 7, 25, 13, 0) first_item = _item(id=1, good_qty=192, started_at=start_at) second_item = _item(id=2, good_qty=48, started_at=second_started_at) report = _report(start_at, end_at, [first_item, second_item], devices=[_device(start_at)]) metrics = calculate_report_metrics(start_at, end_at, [first_item, second_item], devices=[_device(start_at)]) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 240 assert metrics["shift_day_minutes"] == 240 assert {first_item.id: first_item.allocated_minutes, second_item.id: second_item.allocated_minutes} == { 1: 192, 2: 48, } assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [ (1, 192, 192), (2, 48, 48), ] assert sum(row.effective_minutes for row in rows) == 240 def test_matched_meal_split_and_unmatched_device_range_follow_metrics_order(): start_at = datetime(2026, 7, 25, 8, 0) second_started_at = datetime(2026, 7, 25, 12, 0) switch_at = datetime(2026, 7, 25, 13, 0) end_at = datetime(2026, 7, 25, 20, 0) devices = [ _device(start_at, device_no="23#", sort_order=0), _device(switch_at, device_no="77#", sort_order=1), ] first_item = _item(id=1, good_qty=240, device_no="23#", started_at=start_at) second_item = _item(id=2, good_qty=60, device_no="23#", started_at=second_started_at) unmatched_item = _item(id=3, good_qty=320, device_no="99#") items = [first_item, second_item, unmatched_item] report = _report(start_at, end_at, items, devices=devices) metrics = calculate_report_metrics(start_at, end_at, items, devices=devices) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 620 assert {item.id: item.allocated_minutes for item in items} == { 1: 240, 2: 60, 3: 320, } assert { item_id: round(sum(row.effective_minutes for row in rows if row.report_item_id == item_id), 2) for item_id in [1, 2, 3] } == { 1: 240, 2: 60, 3: 320, } assert round(sum(row.effective_minutes for row in rows), 2) == metrics["effective_minutes"] def test_item_totals_anchor_to_metrics_when_unmatched_remaining_buckets_exceed_item_target(): start_at = datetime(2026, 7, 25, 8, 0) second_started_at = datetime(2026, 7, 25, 12, 0) switch_at = datetime(2026, 7, 25, 18, 0) end_at = datetime(2026, 7, 25, 20, 0) devices = [ _device(start_at, device_no="23#", sort_order=0), _device(switch_at, device_no="77#", sort_order=1), ] first_item = _item(id=1, good_qty=240, device_no="23#", started_at=start_at) second_item = _item(id=2, good_qty=360, device_no="23#", started_at=second_started_at) unmatched_item = _item(id=3, good_qty=60, device_no="99#") items = [first_item, second_item, unmatched_item] report = _report(start_at, end_at, items, devices=devices) metrics = calculate_report_metrics(start_at, end_at, items, devices=devices) rows = build_report_allocation_drafts(report) assert metrics["effective_minutes"] == 660 assert {item.id: item.allocated_minutes for item in items} == { 1: 240, 2: 360, 3: 60, } assert { item_id: round(sum(row.effective_minutes for row in rows if row.report_item_id == item_id), 2) for item_id in [1, 2, 3] } == { 1: 240, 2: 360, 3: 60, } assert { "day": round(sum(row.day_minutes for row in rows), 2), "overtime": round(sum(row.overtime_minutes for row in rows), 2), "night": round(sum(row.night_minutes for row in rows), 2), } == { "day": metrics["shift_day_minutes"], "overtime": metrics["shift_overtime_minutes"], "night": metrics["shift_night_minutes"], } assert round(sum(row.effective_minutes for row in rows), 2) == 660 assert all(row.effective_minutes == row.day_minutes + row.overtime_minutes + row.night_minutes for row in rows) def test_zero_effective_minutes_falls_back_to_original_report_date(): item = _item(good_qty=5, defect_qty=1, scrap_qty=0.5, changeover_count=2, process_unit_price_yuan=3) report = _report( datetime(2026, 7, 25, 9, 0), datetime(2026, 7, 25, 9, 0), item, report_date=date(2026, 7, 25), ) rows = build_report_allocation_drafts(report) assert len(rows) == 1 row = rows[0] assert row.allocation_date == date(2026, 7, 25) assert row.effective_minutes == 0 assert row.day_minutes == 0 assert row.overtime_minutes == 0 assert row.night_minutes == 0 assert row.good_qty == 5 assert row.defect_qty == 1 assert row.scrap_qty == 0.5 assert row.changeover_count == 2 assert row.reference_wage == 15 def test_allocation_summary_text_groups_dates_and_shift_kinds(): item = _item(good_qty=650, process_unit_price_yuan=1) report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) rows = build_report_allocation_drafts(report) assert allocation_summary_text(rows) == "2026-07-24 夜班5.83小时；2026-07-25 白班3小时、加班2小时" def test_report_out_exposes_allocation_summary_and_item_allocations(): report = ProductionReport( id=1, session_id=1, attendance_point_name="总厂", employee_phone="13800000000", report_date=date(2026, 7, 25), start_at=datetime(2026, 7, 25, 8, 0), end_at=datetime(2026, 7, 25, 18, 30), break_minutes=60, status=ReportStatus.pending, submitted_at=datetime(2026, 7, 25, 18, 31), ) report.employee = Personnel(phone="13800000000", name="张三") report.session = WorkSession( id=1, employee_phone="13800000000", attendance_point_name="总厂", start_at=datetime(2026, 7, 25, 8, 0), status=SessionStatus.submitted, ) report.session.devices = [] item = ProductionReportItem( id=1, report_id=1, attendance_point_name="总厂", device_no="23#", project_no="P1", product_name="23#", process_name="冲压", process_unit_price_yuan=Decimal("2.50"), good_qty=Decimal("100.50"), defect_qty=Decimal("2.25"), scrap_qty=Decimal("1.25"), changeover_count=Decimal("1.50"), standard_beat=Decimal("3.00"), standard_workload=Decimal("0.00"), ) allocation = ProductionReportAllocation( id=1, report_id=1, report_item_id=1, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 25), day_minutes=Decimal("120.50"), overtime_minutes=Decimal("30.25"), night_minutes=Decimal("0.00"), effective_minutes=Decimal("150.75"), good_qty=Decimal("100.50"), defect_qty=Decimal("2.25"), scrap_qty=Decimal("1.25"), changeover_count=Decimal("1.50"), reference_wage=Decimal("251.25"), ) previous_allocation = ProductionReportAllocation( id=2, report_id=1, report_item_id=1, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 24), day_minutes=Decimal("0.00"), overtime_minutes=Decimal("0.00"), night_minutes=Decimal("60.00"), effective_minutes=Decimal("60.00"), good_qty=Decimal("40.00"), defect_qty=Decimal("0.00"), scrap_qty=Decimal("0.00"), changeover_count=Decimal("0.00"), reference_wage=Decimal("100.00"), ) item.allocations = [allocation, previous_allocation] report.items = [item] report.allocations = [allocation, previous_allocation] report.audit_logs = [] output = report_out(report) assert output.allocation_summary_text == "2026-07-24 夜班1小时；2026-07-25 白班2.01小时、加班0.5小时" assert [row.allocation_date for row in output.items[0].allocations] == [date(2026, 7, 24), date(2026, 7, 25)] item_allocation = output.items[0].allocations[1] assert item_allocation.allocation_date == date(2026, 7, 25) assert item_allocation.day_minutes == 120.5 assert item_allocation.overtime_minutes == 30.25 assert item_allocation.effective_minutes == 150.75 assert item_allocation.good_qty == 100.5 assert item_allocation.defect_qty == 2.25 assert item_allocation.scrap_qty == 1.25 assert item_allocation.changeover_count == 1.5 assert item_allocation.reference_wage == 251.25 def test_report_out_handles_empty_allocations(): report = ProductionReport( id=1, session_id=1, attendance_point_name="总厂", employee_phone="13800000000", report_date=date(2026, 7, 25), start_at=datetime(2026, 7, 25, 8, 0), end_at=datetime(2026, 7, 25, 9, 0), break_minutes=0, status=ReportStatus.pending, submitted_at=datetime(2026, 7, 25, 9, 1), ) report.employee = Personnel(phone="13800000000", name="张三") report.session = WorkSession( id=1, employee_phone="13800000000", attendance_point_name="总厂", start_at=datetime(2026, 7, 25, 8, 0), status=SessionStatus.submitted, ) report.session.devices = [] item = ProductionReportItem( id=1, report_id=1, attendance_point_name="总厂", device_no="23#", project_no="P1", product_name="23#", process_name="冲压", process_unit_price_yuan=0, good_qty=0, defect_qty=0, scrap_qty=0, changeover_count=0, standard_beat=1, standard_workload=0, ) item.allocations = [] report.items = [item] report.allocations = [] report.audit_logs = [] output = report_out(report) assert output.allocation_summary_text == "" assert output.items[0].allocations == [] def test_refresh_report_allocations_replaces_existing_rows_and_flushes(): engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) SessionLocal = sessionmaker(bind=engine, future=True) db = SessionLocal() try: report = ProductionReport( id=1, session_id=1, attendance_point_name="总厂", employee_phone="13800000000", report_date=date(2026, 7, 25), start_at=datetime(2026, 7, 25, 5, 59), end_at=datetime(2026, 7, 25, 11, 0), ) item = ProductionReportItem( id=1, report_id=1, attendance_point_name="总厂", device_no="23#", project_no="P1", product_name="23#", process_name="冲压", process_unit_price_yuan=2, good_qty=301, defect_qty=0, scrap_qty=0, changeover_count=0, standard_beat=1, standard_workload=0, ) stale = ProductionReportAllocation( id=99, report_id=1, report_item_id=1, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 20), effective_minutes=1, ) report.items.append(item) db.add_all([report, stale]) db.flush() rows = refresh_report_allocations(db, report) assert len(rows) == 2 assert all(row.id is not None for row in rows) stored = db.scalars(select(ProductionReportAllocation).order_by(ProductionReportAllocation.allocation_date)).all() assert [row.allocation_date for row in stored] == [date(2026, 7, 24), date(2026, 7, 25)] assert [row.good_qty for row in stored] == [1, 300] finally: db.close() def _sqlite_db(): engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) SessionLocal = sessionmaker(bind=engine, future=True) return SessionLocal() def test_report_allocation_schema_has_unique_report_item_date_key(): engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) inspector = inspect(engine) unique_column_sets = [ tuple(constraint["column_names"]) for constraint in inspector.get_unique_constraints("production_report_allocations") ] unique_column_sets.extend( tuple(index["column_names"]) for index in inspector.get_indexes("production_report_allocations") if index.get("unique") ) assert ("report_id", "report_item_id", "allocation_date") in unique_column_sets def test_report_allocation_rejects_duplicate_report_item_date_key(): db = _sqlite_db() try: first = ProductionReportAllocation( report_id=1, report_item_id=1, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 25), ) duplicate = ProductionReportAllocation( report_id=1, report_item_id=1, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 25), ) db.add(first) db.flush() db.add(duplicate) with pytest.raises(IntegrityError): db.flush() finally: db.close() def test_report_out_skips_unloaded_allocations_without_lazy_load(): db = _sqlite_db() try: person = Personnel(phone="13800000000", name="张三") session = WorkSession( id=1, attendance_point_name="总厂", employee_phone=person.phone, start_at=datetime(2026, 7, 25, 8, 0), end_at=datetime(2026, 7, 25, 10, 0), status=SessionStatus.submitted, ) report = ProductionReport( id=1, session_id=session.id, attendance_point_name="总厂", employee_phone=person.phone, report_date=date(2026, 7, 25), start_at=datetime(2026, 7, 25, 8, 0), end_at=datetime(2026, 7, 25, 10, 0), break_minutes=0, status=ReportStatus.pending, submitted_at=datetime(2026, 7, 25, 10, 1), ) item = ProductionReportItem( id=1, report_id=report.id, attendance_point_name="总厂", device_no="23#", project_no="P1", product_name="23#", process_name="冲压", process_unit_price_yuan=1, good_qty=100, defect_qty=0, scrap_qty=0, changeover_count=0, standard_beat=1, standard_workload=0, ) allocation = ProductionReportAllocation( id=1, report_id=report.id, report_item_id=item.id, attendance_point_name="总厂", employee_phone=person.phone, allocation_date=date(2026, 7, 25), day_minutes=120, effective_minutes=120, good_qty=100, reference_wage=100, ) db.add_all([person, session, report, item, allocation]) db.commit() loaded_report = db.scalar( select(ProductionReport) .options( selectinload(ProductionReport.employee), selectinload(ProductionReport.items).raiseload(ProductionReportItem.allocations), selectinload(ProductionReport.audit_logs), selectinload(ProductionReport.session).selectinload(WorkSession.devices), raiseload(ProductionReport.allocations), ) .where(ProductionReport.id == report.id) ) output = report_out(loaded_report) assert output.allocation_summary_text == "" assert output.items[0].allocations == [] finally: db.close() def _product( *, point_name: str = "总厂", project_no: str = "P1", product_name: str = "23#", process_name: str = "冲压", stamping_method: str | None = None, ): return Product( attendance_point_name=point_name, project_no=project_no, product_name=product_name, device_no="", process_name=process_name, stamping_method=stamping_method, operator_count=1, process_unit_price_yuan=2, standard_beat=1, standard_workload=0, ) def _press_equipment(point_name: str = "总厂", device_no: str = "23#"): return Equipment( attendance_point_name=point_name, device_no=device_no, device_type="冲压设备", ) def _session_with_device( *, point_name: str = "总厂", employee_phone: str = "13800000000", start_at: datetime = datetime(2026, 7, 25, 8, 0), end_at: datetime = datetime(2026, 7, 25, 11, 0), ): return WorkSession( attendance_point_name=point_name, employee_phone=employee_phone, start_at=start_at, end_at=end_at, status=SessionStatus.reporting, devices=[ WorkSessionDevice( attendance_point_name=point_name, device_no="23#", process_name="冲压", scanned_at=start_at, sort_order=0, ) ], ) def _stored_allocations(db, report_id: int): return db.scalars( select(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report_id) ).all() def test_submit_report_creates_allocations(monkeypatch): db = _sqlite_db() try: user = SimpleNamespace(phone="13800000000") session = _session_with_device(employee_phone=user.phone) db.add_all([session, _press_equipment(), _product()]) db.commit() monkeypatch.setattr(reports_router, "report_out", lambda report, schedule=None: report) monkeypatch.setattr(reporting_window_service, "now", lambda: datetime(2026, 7, 25, 11, 5)) report = reports_router.submit_report( ReportSubmitRequest( session_id=session.id, device_blocks=[ ReportSubmitBlock( device_no="23#", process_name="冲压", items=[ ReportSubmitItem( device_no="23#", project_no="P1", product_name="23#", process_name="冲压", standard_beat=1, good_qty=180, ) ], ) ], ), user=user, db=db, ) allocations = _stored_allocations(db, report.id) assert len(allocations) == 1 assert allocations[0].report_item_id == report.items[0].id assert allocations[0].effective_minutes == 180 assert allocations[0].good_qty == 180 finally: db.close() def test_submit_cleaning_report_creates_zero_minute_allocation(monkeypatch): db = _sqlite_db() try: point_name = "总厂" user = SimpleNamespace(phone="13800000000", role=Role.manager) db.add_all( [ AttendancePoint(name=point_name, latitude=0, longitude=0, radius_meters=500, is_active=True), Equipment(attendance_point_name=point_name, device_no="C1", device_type="清洗设备"), _product( point_name=point_name, project_no="CLEAN1", product_name="清洗产品", process_name="清洗", stamping_method="清洗", ), ] ) db.commit() monkeypatch.setattr(reports_router, "report_out", lambda report, schedule=None: report) report = reports_router.submit_cleaning_report( CleaningReportSubmitRequest( latitude=0, longitude=0, items=[ CleaningReportSubmitItem( attendance_point_name=point_name, device_no="C1", product_name="清洗产品", process_name="清洗", quantity=25, ) ], ), user=user, db=db, ) allocations = _stored_allocations(db, report.id) assert len(allocations) == 1 assert allocations[0].effective_minutes == 0 assert allocations[0].good_qty == 25 finally: db.close() def test_auto_submit_session_creates_allocations(): db = _sqlite_db() try: session = _session_with_device( start_at=datetime(2026, 7, 25, 8, 0), end_at=None, ) session.status = SessionStatus.active db.add_all([session, _product()]) db.commit() report = auto_submit_service.auto_submit_session( db, session, submitted_at=datetime(2026, 7, 25, 11, 0), ) db.flush() allocations = _stored_allocations(db, report.id) assert len(allocations) == 1 assert allocations[0].report_item_id == report.items[0].id assert allocations[0].effective_minutes == 180 finally: db.close() def test_approve_report_refreshes_allocations_after_edit_save(monkeypatch): db = _sqlite_db() try: user = SimpleNamespace(phone="13900000000", role=Role.admin) session = _session_with_device() session.status = SessionStatus.submitted report = ProductionReport( attendance_point_name="总厂", session=session, employee_phone="13800000000", report_date=date(2026, 7, 25), start_at=datetime(2026, 7, 25, 8, 0), end_at=datetime(2026, 7, 25, 11, 0), duration_minutes=180, effective_minutes=180, total_good_qty=100, total_output_qty=100, actual_beat=1.8, standard_beat=1, expected_workload=100, pace_rate=100, workload_rate=100, status=ReportStatus.pending, submitted_at=datetime(2026, 7, 25, 11, 1), items=[ ProductionReportItem( attendance_point_name="总厂", device_no="23#", project_no="P1", product_name="23#", process_name="冲压", process_unit_price_yuan=2, standard_beat=1, standard_workload=0, good_qty=100, defect_qty=0, scrap_qty=0, ) ], ) db.add_all([_press_equipment(), _product(), report]) db.flush() db.add( ProductionReportAllocation( report_id=report.id, report_item_id=report.items[0].id, attendance_point_name="总厂", employee_phone="13800000000", allocation_date=date(2026, 7, 20), effective_minutes=1, good_qty=1, ) ) db.commit() monkeypatch.setattr(reviews_router, "accessible_point_names", lambda _db, _user: ["总厂"]) monkeypatch.setattr(reviews_router, "report_out", lambda report, schedule=None: report) saved = reviews_router.approve_report( report.id, ApproveReportRequest( item_corrections=[ ReportItemCorrection(id=report.items[0].id, good_qty=200) ] ), user=user, db=db, ) allocations = _stored_allocations(db, saved.id) assert len(allocations) == 1 assert allocations[0].allocation_date == date(2026, 7, 25) assert allocations[0].good_qty == 200 assert allocations[0].reference_wage == 400 finally: db.close() def test_purge_expired_voided_reports_deletes_allocations_before_report_items(): report = SimpleNamespace(id=1, session=None) operations: list[str] = [] class _Rows: def all(self): return [report] class _FakeDb: def scalars(self, _statement): return _Rows() def execute(self, statement): operations.append(statement.table.name) def delete(self, _obj): operations.append("delete-report") def commit(self): operations.append("commit") purged_count = purge_expired_voided_reports(_FakeDb()) assert purged_count == 1 assert operations.index("production_report_allocations") < operations.index("production_report_items")