283 lines
9.1 KiB
Python
283 lines
9.1 KiB
Python
from datetime import datetime, timedelta
|
||
|
||
from app.services.metrics import calculate_report_metrics
|
||
from app.services.work_schedule import WorkScheduleConfig
|
||
|
||
|
||
class Item:
|
||
def __init__(
|
||
self,
|
||
device_no: str = "23#",
|
||
good_qty: float = 2,
|
||
standard_beat: float = 21.12,
|
||
standard_workload: float = 1513,
|
||
started_at: datetime | None = None,
|
||
) -> None:
|
||
self.device_no = device_no
|
||
self.started_at = started_at
|
||
self.standard_beat = standard_beat
|
||
self.standard_workload = standard_workload
|
||
self.good_qty = good_qty
|
||
self.defect_qty = 0
|
||
self.scrap_qty = 0
|
||
self.allocated_minutes = 0
|
||
|
||
|
||
class Device:
|
||
def __init__(
|
||
self,
|
||
device_no: str,
|
||
scanned_at: datetime,
|
||
sort_order: int,
|
||
released_at: datetime | None = None,
|
||
release_reason: str | None = None,
|
||
) -> None:
|
||
self.device_no = device_no
|
||
self.scanned_at = scanned_at
|
||
self.sort_order = sort_order
|
||
self.released_at = released_at
|
||
self.release_reason = release_reason
|
||
|
||
|
||
def test_short_report_uses_actual_seconds_without_minute_floor():
|
||
start_at = datetime(2026, 5, 3, 1, 4, 50)
|
||
end_at = start_at + timedelta(seconds=2)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [Item()])
|
||
|
||
assert metrics["duration_minutes"] == 0.03
|
||
assert metrics["effective_minutes"] == 0.03
|
||
assert metrics["actual_beat"] == 1.0
|
||
|
||
|
||
def test_multi_device_report_allocates_minutes_by_switch_time():
|
||
start_at = datetime(2026, 5, 3, 1, 25, 55)
|
||
switch_at = start_at + timedelta(seconds=30)
|
||
end_at = switch_at + timedelta(seconds=30)
|
||
first_item = Item(device_no="23#", good_qty=3)
|
||
second_item = Item(device_no="28#", good_qty=6)
|
||
|
||
metrics = calculate_report_metrics(
|
||
start_at,
|
||
end_at,
|
||
[first_item, second_item],
|
||
devices=[
|
||
Device("23#", start_at, 0),
|
||
Device("28#", switch_at, 1),
|
||
],
|
||
)
|
||
|
||
assert first_item.allocated_minutes == 0.5
|
||
assert second_item.allocated_minutes == 0.5
|
||
assert first_item.allocated_minutes * 60 / first_item.good_qty == 10
|
||
assert second_item.allocated_minutes * 60 / second_item.good_qty == 5
|
||
assert metrics["duration_minutes"] == 1
|
||
|
||
|
||
def test_multi_device_report_last_segment_uses_report_end_time():
|
||
start_at = datetime(2026, 5, 3, 1, 56, 57)
|
||
switch_at = start_at + timedelta(seconds=30)
|
||
end_at = start_at + timedelta(seconds=44)
|
||
first_item = Item(device_no="23#", good_qty=3)
|
||
second_item = Item(device_no="28#", good_qty=5)
|
||
|
||
metrics = calculate_report_metrics(
|
||
start_at,
|
||
end_at,
|
||
[first_item, second_item],
|
||
devices=[
|
||
Device("23#", start_at, 0),
|
||
Device("28#", switch_at, 1),
|
||
],
|
||
)
|
||
|
||
total_allocated_seconds = round((first_item.allocated_minutes + second_item.allocated_minutes) * 60)
|
||
assert metrics["duration_minutes"] == 0.73
|
||
assert round(first_item.allocated_minutes * 60) == 30
|
||
assert round(second_item.allocated_minutes * 60) == 14
|
||
assert total_allocated_seconds == 44
|
||
|
||
|
||
def test_same_device_items_split_by_item_start_time():
|
||
start_at = datetime(2026, 5, 3, 21, 51, 35)
|
||
product_switch_at = start_at + timedelta(seconds=20)
|
||
end_at = start_at + timedelta(seconds=50)
|
||
first_item = Item(device_no="28#", good_qty=1, started_at=start_at)
|
||
second_item = Item(device_no="28#", good_qty=3, started_at=product_switch_at)
|
||
|
||
metrics = calculate_report_metrics(
|
||
start_at,
|
||
end_at,
|
||
[first_item, second_item],
|
||
devices=[
|
||
Device("28#", start_at, 0),
|
||
],
|
||
)
|
||
|
||
assert round(first_item.allocated_minutes * 60) == 20
|
||
assert round(second_item.allocated_minutes * 60) == 30
|
||
assert metrics["duration_minutes"] == 0.83
|
||
|
||
|
||
def test_meal_break_is_deducted_and_shift_distribution_is_marked():
|
||
start_at = datetime(2026, 5, 4, 8, 0, 0)
|
||
end_at = datetime(2026, 5, 4, 20, 0, 0)
|
||
item = Item(good_qty=10)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [item])
|
||
|
||
assert metrics["duration_minutes"] == 720
|
||
assert metrics["shift_meal_break_minutes"] == 100
|
||
assert metrics["effective_minutes"] == 620
|
||
assert metrics["shift_day_minutes"] == 500
|
||
assert metrics["shift_overtime_minutes"] == 120
|
||
assert metrics["shift_distribution_text"] == "工时 10.33小时（白班8.33小时、加班2小时）"
|
||
|
||
|
||
def test_partial_meal_tail_is_not_deducted_when_report_ends_during_meal():
|
||
schedule = WorkScheduleConfig(
|
||
day_start="08:00",
|
||
day_end="17:00",
|
||
lunch_start="11:40",
|
||
lunch_end="12:40",
|
||
dinner_start="17:00",
|
||
dinner_end="17:30",
|
||
overtime_start="17:00",
|
||
overtime_end="20:00",
|
||
night_start="20:00",
|
||
night_end="06:00",
|
||
)
|
||
start_at = datetime(2026, 5, 4, 16, 0, 0)
|
||
end_at = datetime(2026, 5, 4, 17, 10, 0)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [Item()], schedule=schedule)
|
||
|
||
assert metrics["duration_minutes"] == 70
|
||
assert metrics["shift_meal_break_minutes"] == 0
|
||
assert metrics["effective_minutes"] == 70
|
||
assert metrics["shift_day_minutes"] == 60
|
||
assert metrics["shift_overtime_minutes"] == 10
|
||
|
||
|
||
def test_meal_end_grace_keeps_minutes_until_five_minutes_after_meal():
|
||
schedule = WorkScheduleConfig(
|
||
day_start="08:00",
|
||
day_end="17:00",
|
||
lunch_start="11:40",
|
||
lunch_end="12:40",
|
||
dinner_start="17:00",
|
||
dinner_end="17:30",
|
||
overtime_start="17:00",
|
||
overtime_end="20:00",
|
||
night_start="20:00",
|
||
night_end="06:00",
|
||
)
|
||
start_at = datetime(2026, 5, 4, 16, 0, 0)
|
||
end_at = datetime(2026, 5, 4, 17, 35, 0)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [Item()], schedule=schedule)
|
||
|
||
assert metrics["duration_minutes"] == 95
|
||
assert metrics["shift_meal_break_minutes"] == 0
|
||
assert metrics["effective_minutes"] == 95
|
||
assert metrics["shift_day_minutes"] == 60
|
||
assert metrics["shift_overtime_minutes"] == 35
|
||
|
||
|
||
def test_meal_break_is_deducted_after_grace_window():
|
||
schedule = WorkScheduleConfig(
|
||
day_start="08:00",
|
||
day_end="17:00",
|
||
lunch_start="11:40",
|
||
lunch_end="12:40",
|
||
dinner_start="17:00",
|
||
dinner_end="17:30",
|
||
overtime_start="17:00",
|
||
overtime_end="20:00",
|
||
night_start="20:00",
|
||
night_end="06:00",
|
||
)
|
||
start_at = datetime(2026, 5, 4, 16, 0, 0)
|
||
end_at = datetime(2026, 5, 4, 17, 36, 0)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [Item()], schedule=schedule)
|
||
|
||
assert metrics["duration_minutes"] == 96
|
||
assert metrics["shift_meal_break_minutes"] == 30
|
||
assert metrics["effective_minutes"] == 66
|
||
assert metrics["shift_day_minutes"] == 60
|
||
assert metrics["shift_overtime_minutes"] == 6
|
||
|
||
|
||
def test_full_meal_break_is_deducted_when_report_covers_whole_meal():
|
||
schedule = WorkScheduleConfig(
|
||
day_start="08:00",
|
||
day_end="17:00",
|
||
lunch_start="11:40",
|
||
lunch_end="12:40",
|
||
dinner_start="17:00",
|
||
dinner_end="17:30",
|
||
overtime_start="17:00",
|
||
overtime_end="20:00",
|
||
night_start="20:00",
|
||
night_end="06:00",
|
||
)
|
||
start_at = datetime(2026, 5, 4, 16, 0, 0)
|
||
end_at = datetime(2026, 5, 4, 18, 0, 0)
|
||
|
||
metrics = calculate_report_metrics(start_at, end_at, [Item()], schedule=schedule)
|
||
|
||
assert metrics["duration_minutes"] == 120
|
||
assert metrics["shift_meal_break_minutes"] == 30
|
||
assert metrics["effective_minutes"] == 90
|
||
assert metrics["shift_day_minutes"] == 60
|
||
assert metrics["shift_overtime_minutes"] == 30
|
||
|
||
|
||
def test_admin_corrected_start_uses_first_device_from_report_start():
|
||
start_at = datetime(2026, 5, 27, 10, 0, 0)
|
||
scanned_at = datetime(2026, 5, 27, 13, 56, 11)
|
||
submitted_at = datetime(2026, 5, 27, 13, 56, 53)
|
||
end_at = datetime(2026, 5, 27, 13, 59, 0)
|
||
item = Item(device_no="开口挡圈-3528221-05", good_qty=2700, standard_beat=1, started_at=scanned_at)
|
||
|
||
metrics = calculate_report_metrics(
|
||
start_at,
|
||
end_at,
|
||
[item],
|
||
devices=[
|
||
Device(
|
||
"开口挡圈-3528221-05",
|
||
scanned_at,
|
||
0,
|
||
released_at=submitted_at,
|
||
release_reason="报工提交释放占用",
|
||
),
|
||
],
|
||
)
|
||
|
||
assert metrics["duration_minutes"] == 239
|
||
assert metrics["effective_minutes"] == 179
|
||
assert item.allocated_minutes == 179
|
||
assert metrics["shift_distribution_text"] == "工时 2.98小时（白班2.98小时）"
|
||
|
||
|
||
def test_admin_forced_release_still_caps_work_time():
|
||
start_at = datetime(2026, 5, 27, 8, 0, 0)
|
||
release_at = datetime(2026, 5, 27, 9, 0, 0)
|
||
end_at = datetime(2026, 5, 27, 12, 0, 0)
|
||
item = Item(device_no="A", good_qty=10)
|
||
|
||
metrics = calculate_report_metrics(
|
||
start_at,
|
||
end_at,
|
||
[item],
|
||
devices=[
|
||
Device("A", start_at, 0, released_at=release_at, release_reason="管理员强制解除占用"),
|
||
],
|
||
)
|
||
|
||
assert metrics["duration_minutes"] == 60
|
||
assert metrics["effective_minutes"] == 60
|
||
assert item.allocated_minutes == 60
|