1110 lines
37 KiB
Python
1110 lines
37 KiB
Python
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")
|