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

fix: 完善报工归属输出安全性

Browse Source
This commit is contained in:
souplearn
2026-07-25 04:39:25 +08:00
parent b0cb217ce3
commit bb9ce5601c
4 changed files with 239 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_dashboard_source_dates.py
+105
View File
@@ -0,0 +1,105 @@
from datetime import date, datetime
from sqlalchemy import BigInteger, create_engine
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import (
AttendancePoint,
Personnel,
Product,
ProductionReport,
ProductionReportItem,
ReportStatus,
SessionStatus,
WorkSession,
)
from app.routers.dashboard import _dashboard_rows
@compiles(BigInteger, "sqlite")
def _compile_big_integer_for_sqlite(type_, compiler, **kw):
return "INTEGER"
def _db_session():
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine, future=True)
return SessionLocal()
def test_dashboard_rows_populate_source_report_dates_from_original_reports():
db = _db_session()
try:
point_name = "总厂"
report_date = date(2026, 7, 25)
person = Personnel(phone="13800000000", name="张三")
product = Product(
attendance_point_name=point_name,
project_no="P1",
product_name="23#",
device_no="",
process_name="冲压",
operator_count=1,
process_unit_price_yuan=2,
standard_beat=1,
standard_workload=0,
)
db.add_all([
AttendancePoint(name=point_name),
person,
product,
])
for index, start_hour in enumerate([8, 10], start=1):
session = WorkSession(
id=index,
attendance_point_name=point_name,
employee_phone=person.phone,
start_at=datetime(2026, 7, 25, start_hour, 0),
end_at=datetime(2026, 7, 25, start_hour + 1, 0),
status=SessionStatus.submitted,
)
report = ProductionReport(
id=index,
session_id=session.id,
attendance_point_name=point_name,
employee_phone=person.phone,
report_date=report_date,
start_at=session.start_at,
end_at=session.end_at,
duration_minutes=60,
break_minutes=0,
effective_minutes=60,
status=ReportStatus.approved,
submitted_at=session.end_at,
)
item = ProductionReportItem(
id=index,
report_id=report.id,
attendance_point_name=point_name,
device_no="23#",
project_no=product.project_no,
product_name=product.product_name,
process_name=product.process_name,
operator_count=1,
process_unit_price_yuan=2,
standard_beat=1,
standard_workload=0,
good_qty=10,
defect_qty=0,
scrap_qty=0,
allocated_minutes=60,
)
db.add_all([session, report, item])
db.commit()
rows = _dashboard_rows(db, report_date, report_date)
assert len(rows) == 1
assert rows[0].report_count == 2
assert rows[0].source_report_date == report_date
assert rows[0].source_report_dates == [report_date]
finally:
db.close()
tests/test_report_allocations.py
+98 -6
View File
@@ -4,7 +4,7 @@ from types import SimpleNamespace
from sqlalchemy import BigInteger, create_engine, select
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import raiseload, selectinload, sessionmaker
from app.database import Base
from app.models import (
@@ -511,16 +511,33 @@ def test_report_out_exposes_allocation_summary_and_item_allocations():
changeover_count=Decimal("1.50"),
reference_wage=Decimal("251.25"),
)
item.allocations = [allocation]
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]
report.allocations = [allocation, previous_allocation]
report.audit_logs = []
output = report_out(report)
assert output.allocation_summary_text == "2026-07-25 白班2.01小时、加班0.5小时"
assert len(output.items[0].allocations) == 1
item_allocation = output.items[0].allocations[0]
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
@@ -643,6 +660,81 @@ def _sqlite_db():
return SessionLocal()
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 = "总厂",