niuniu/JhHardwareWRS_BackPoint
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 增加报工超额校验服务

Browse Source
This commit is contained in:
souplearn
2026-07-25 20:27:00 +08:00
parent 58ff209d0a
commit 1ed48314be
2 changed files with 845 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_report_over_limit.py
+460 -1
View File
@@ -1,8 +1,21 @@
from datetime import datetime
from decimal import Decimal
from types import SimpleNamespace
from sqlalchemy import BigInteger, create_engine, inspect
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import ProductionReport, ProductionReportItem # noqa: F401
from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
from app.services.report_over_limit import (
OVER_LIMIT_STATUS_EXCEEDED,
OVER_LIMIT_STATUS_NOT_CHECKED,
OVER_LIMIT_TYPE_UNCHECKABLE,
current_report_qty,
evaluate_report_item_over_limit,
parse_process_order,
)
@compiles(BigInteger, "sqlite")
@@ -56,3 +69,449 @@ def test_report_over_limit_columns_exist_on_sqlite_metadata():
assert str(report_column_by_name["has_over_limit"]["default"]).strip("'\"") in {"0", "false", "FALSE"}
assert item_column_by_name["over_limit_status"]["nullable"] is False
assert str(item_column_by_name["over_limit_status"]["default"]).strip("'\"") == "not_checked"
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_parse_process_order_accepts_numeric_process_names():
assert parse_process_order("1") == 1
assert parse_process_order("1序") == 1
assert parse_process_order(" 02 序 ") == 2
def test_parse_process_order_rejects_special_and_text_process_names():
assert parse_process_order("杂活") is None
assert parse_process_order("清洗") is None
assert parse_process_order("第一序") is None
def test_current_report_qty_uses_good_defect_and_scrap():
item = SimpleNamespace(good_qty=Decimal("90"), defect_qty=Decimal("5"), scrap_qty=Decimal("5"))
assert current_report_qty(item) == Decimal("100.00")
def _report(report_id: int, status: ReportStatus = ReportStatus.pending) -> ProductionReport:
return ProductionReport(
id=report_id,
session_id=report_id,
attendance_point_name="总厂",
employee_phone="13800000000",
report_date=datetime(2026, 7, 25).date(),
start_at=datetime(2026, 7, 25, 8),
end_at=datetime(2026, 7, 25, 10),
status=status,
)
def test_first_process_exceeds_material_limit_with_submit_snapshot(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
report = _report(1)
item = ProductionReportItem(
id=1,
report=report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=60,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, report, item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
assert result.max_reportable_qty == Decimal("50.00")
assert result.current_report_qty == Decimal("60.00")
assert result.current_cumulative_qty == Decimal("60.00")
assert "1序累计报工数量超过本批次材料可支撑数量" in result.reason
finally:
db.close()
def test_evaluate_defaults_to_excluding_current_report(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
report = _report(1)
item = ProductionReportItem(
id=1,
report=report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=50,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, report, item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, item)
assert result.current_cumulative_qty == Decimal("50.00")
assert result.status != OVER_LIMIT_STATUS_EXCEEDED
finally:
db.close()
def test_current_report_sibling_items_contribute_to_current_cumulative(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
report = _report(1)
first_item = ProductionReportItem(
id=1,
report=report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
second_item = ProductionReportItem(
id=2,
report=report,
attendance_point_name="总厂",
device_no="B",
project_no="P1",
product_name="产品A",
raw_material_batch_no=" LOT-1 ",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, report, first_item, second_item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, first_item)
assert result.current_report_qty == Decimal("30.00")
assert result.current_cumulative_qty == Decimal("60.00")
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
finally:
db.close()
def test_history_batch_numbers_are_compared_after_trim(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
previous_report = _report(1, ReportStatus.approved)
previous_item = ProductionReportItem(
id=1,
report=previous_report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no=" LOT-1 ",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
report = _report(2)
item = ProductionReportItem(
id=2,
report=report,
attendance_point_name="总厂",
device_no="B",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, previous_report, previous_item, report, item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, item)
assert result.current_cumulative_qty == Decimal("60.00")
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
finally:
db.close()
def test_erp_weight_lookup_sql_error_has_distinct_uncheckable_reason():
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
report = _report(1)
item = ProductionReportItem(
id=1,
report=report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, report, item])
db.flush()
result = evaluate_report_item_over_limit(db, report, item)
assert result.over_limit_type == OVER_LIMIT_TYPE_UNCHECKABLE
assert result.reason == "读取批次投入重量失败，无法校验超额"
finally:
db.close()
def test_evaluate_report_item_over_limit_does_not_write_snapshot_fields(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
report = _report(1)
item = ProductionReportItem(
id=1,
report=report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=60,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, report, item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, item)
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
assert item.over_limit_status == OVER_LIMIT_STATUS_NOT_CHECKED
assert item.over_limit_type is None
assert item.over_limit_reason is None
finally:
db.close()
def test_later_process_exceeds_previous_good_quantity():
db = _sqlite_db()
try:
db.add_all(
[
Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
standard_beat=1,
),
Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="3",
standard_beat=1,
),
]
)
previous_report = _report(1, ReportStatus.approved)
previous_item = ProductionReportItem(
id=1,
report=previous_report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=100,
defect_qty=10,
scrap_qty=5,
standard_beat=1,
)
report = _report(2)
report.employee_phone = "13800000001"
item = ProductionReportItem(
id=2,
report=report,
attendance_point_name="总厂",
device_no="B",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="3",
good_qty=98,
defect_qty=3,
scrap_qty=2,
standard_beat=1,
)
db.add_all([previous_report, previous_item, report, item])
db.flush()
result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
assert result.previous_good_qty == Decimal("100.00")
assert result.current_report_qty == Decimal("103.00")
assert result.current_cumulative_qty == Decimal("103.00")
assert "3序累计报工数量超过1序累计成品数量" in result.reason
finally:
db.close()
def test_history_process_names_are_compared_after_trim(monkeypatch):
db = _sqlite_db()
try:
product = Product(
attendance_point_name="总厂",
project_no="P1",
product_name="产品A",
device_no="",
process_name="1",
product_gross_weight_kg=Decimal("2"),
standard_beat=1,
)
previous_report = _report(1, ReportStatus.approved)
previous_item = ProductionReportItem(
id=1,
report=previous_report,
attendance_point_name="总厂",
device_no="A",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name=" 01序 ",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
report = _report(2)
item = ProductionReportItem(
id=2,
report=report,
attendance_point_name="总厂",
device_no="B",
project_no="P1",
product_name="产品A",
raw_material_batch_no="LOT-1",
process_name="1",
good_qty=30,
defect_qty=0,
scrap_qty=0,
standard_beat=1,
)
db.add_all([product, previous_report, previous_item, report, item])
db.flush()
monkeypatch.setattr(
"app.services.report_over_limit.read_batch_issued_weight_kg",
lambda db, batch_no, project_no, product_name: Decimal("100"),
)
result = evaluate_report_item_over_limit(db, report, item)
assert result.current_cumulative_qty == Decimal("60.00")
assert result.status == OVER_LIMIT_STATUS_EXCEEDED
finally:
db.close()