refactor: Overlapping validation for Shift Request
- commonify time overlap function between request and assignment
- add tests for shift request overlap
diff --git a/erpnext/hr/doctype/shift_assignment/shift_assignment.py b/erpnext/hr/doctype/shift_assignment/shift_assignment.py
index fd0b4d5..0b21c00 100644
--- a/erpnext/hr/doctype/shift_assignment/shift_assignment.py
+++ b/erpnext/hr/doctype/shift_assignment/shift_assignment.py
@@ -32,7 +32,7 @@
overlapping_dates = self.get_overlapping_dates()
if len(overlapping_dates):
# if dates are overlapping, check if timings are overlapping, else allow
- overlapping_timings = self.has_overlapping_timings(overlapping_dates[0].shift_type)
+ overlapping_timings = has_overlapping_timings(self.shift_type, overlapping_dates[0].shift_type)
if overlapping_timings:
self.throw_overlap_error(overlapping_dates[0])
@@ -43,9 +43,7 @@
shift = frappe.qb.DocType("Shift Assignment")
query = (
frappe.qb.from_(shift)
- .select(
- shift.name, shift.shift_type, shift.start_date, shift.end_date, shift.docstatus, shift.status
- )
+ .select(shift.name, shift.shift_type, shift.docstatus, shift.status)
.where(
(shift.employee == self.employee)
& (shift.docstatus == 1)
@@ -81,55 +79,46 @@
return query.run(as_dict=True)
- def has_overlapping_timings(self, overlapping_shift):
- curr_shift = frappe.db.get_value(
- "Shift Type", self.shift_type, ["start_time", "end_time"], as_dict=True
- )
- overlapping_shift = frappe.db.get_value(
- "Shift Type", overlapping_shift, ["start_time", "end_time"], as_dict=True
- )
-
- if (
- (
- curr_shift.start_time > overlapping_shift.start_time
- and curr_shift.start_time < overlapping_shift.end_time
- )
- or (
- curr_shift.end_time > overlapping_shift.start_time
- and curr_shift.end_time < overlapping_shift.end_time
- )
- or (
- curr_shift.start_time <= overlapping_shift.start_time
- and curr_shift.end_time >= overlapping_shift.end_time
- )
- ):
- return True
- return False
-
def throw_overlap_error(self, shift_details):
shift_details = frappe._dict(shift_details)
- msg = None
if shift_details.docstatus == 1 and shift_details.status == "Active":
- if shift_details.start_date and shift_details.end_date:
- msg = _("Employee {0} already has an active Shift {1}: {2} from {3} to {4}").format(
- frappe.bold(self.employee),
- frappe.bold(self.shift_type),
- get_link_to_form("Shift Assignment", shift_details.name),
- getdate(self.start_date).strftime("%d-%m-%Y"),
- getdate(self.end_date).strftime("%d-%m-%Y"),
- )
- else:
- msg = _("Employee {0} already has an active Shift {1}: {2} from {3}").format(
- frappe.bold(self.employee),
- frappe.bold(self.shift_type),
- get_link_to_form("Shift Assignment", shift_details.name),
- getdate(self.start_date).strftime("%d-%m-%Y"),
- )
-
- if msg:
+ msg = _(
+ "Employee {0} already has an active Shift {1}: {2} that overlaps within this period."
+ ).format(
+ frappe.bold(self.employee),
+ frappe.bold(shift_details.shift_type),
+ get_link_to_form("Shift Assignment", shift_details.name),
+ )
frappe.throw(msg, title=_("Overlapping Shifts"), exc=OverlappingShiftError)
+def has_overlapping_timings(shift_1: str, shift_2: str) -> bool:
+ """
+ Accepts two shift types and checks whether their timings are overlapping
+ """
+ curr_shift = frappe.db.get_value("Shift Type", shift_1, ["start_time", "end_time"], as_dict=True)
+ overlapping_shift = frappe.db.get_value(
+ "Shift Type", shift_2, ["start_time", "end_time"], as_dict=True
+ )
+
+ if (
+ (
+ curr_shift.start_time > overlapping_shift.start_time
+ and curr_shift.start_time < overlapping_shift.end_time
+ )
+ or (
+ curr_shift.end_time > overlapping_shift.start_time
+ and curr_shift.end_time < overlapping_shift.end_time
+ )
+ or (
+ curr_shift.start_time <= overlapping_shift.start_time
+ and curr_shift.end_time >= overlapping_shift.end_time
+ )
+ ):
+ return True
+ return False
+
+
@frappe.whitelist()
def get_events(start, end, filters=None):
events = []
diff --git a/erpnext/hr/doctype/shift_assignment/test_shift_assignment.py b/erpnext/hr/doctype/shift_assignment/test_shift_assignment.py
index 048b573..0fe9108 100644
--- a/erpnext/hr/doctype/shift_assignment/test_shift_assignment.py
+++ b/erpnext/hr/doctype/shift_assignment/test_shift_assignment.py
@@ -103,7 +103,7 @@
# shift with end date
make_shift_assignment(shift_type.name, employee, date, add_days(date, 30))
- # shift setup for 13-15
+ # shift setup for 11-15
shift_type = setup_shift_type(shift_type="Shift 2", start_time="11:00:00", end_time="15:00:00")
date = getdate()
@@ -127,7 +127,7 @@
date = getdate()
make_shift_assignment(shift_type.name, employee, date)
- # shift setup for 13-15
+ # shift setup for 11-15
shift_type = setup_shift_type(shift_type="Shift 2", start_time="11:00:00", end_time="15:00:00")
date = getdate()
diff --git a/erpnext/hr/doctype/shift_request/shift_request.py b/erpnext/hr/doctype/shift_request/shift_request.py
index 1e3e8ff..083aa3d 100644
--- a/erpnext/hr/doctype/shift_request/shift_request.py
+++ b/erpnext/hr/doctype/shift_request/shift_request.py
@@ -5,12 +5,14 @@
import frappe
from frappe import _
from frappe.model.document import Document
-from frappe.utils import formatdate, getdate
+from frappe.query_builder import Criterion
+from frappe.utils import formatdate, get_link_to_form, getdate
+from erpnext.hr.doctype.shift_assignment.shift_assignment import has_overlapping_timings
from erpnext.hr.utils import share_doc_with_approver, validate_active_employee
-class OverlapError(frappe.ValidationError):
+class OverlappingShiftRequestError(frappe.ValidationError):
pass
@@ -18,7 +20,7 @@
def validate(self):
validate_active_employee(self.employee)
self.validate_dates()
- self.validate_shift_request_overlap_dates()
+ self.validate_overlapping_shift_requests()
self.validate_approver()
self.validate_default_shift()
@@ -79,37 +81,60 @@
if self.from_date and self.to_date and (getdate(self.to_date) < getdate(self.from_date)):
frappe.throw(_("To date cannot be before from date"))
- def validate_shift_request_overlap_dates(self):
+ def validate_overlapping_shift_requests(self):
+ overlapping_dates = self.get_overlapping_dates()
+ if len(overlapping_dates):
+ # if dates are overlapping, check if timings are overlapping, else allow
+ overlapping_timings = has_overlapping_timings(self.shift_type, overlapping_dates[0].shift_type)
+ if overlapping_timings:
+ self.throw_overlap_error(overlapping_dates[0])
+
+ def get_overlapping_dates(self):
if not self.name:
self.name = "New Shift Request"
- d = frappe.db.sql(
- """
- select
- name, shift_type, from_date, to_date
- from `tabShift Request`
- where employee = %(employee)s and docstatus < 2
- and ((%(from_date)s >= from_date
- and %(from_date)s <= to_date) or
- ( %(to_date)s >= from_date
- and %(to_date)s <= to_date ))
- and name != %(name)s""",
- {
- "employee": self.employee,
- "shift_type": self.shift_type,
- "from_date": self.from_date,
- "to_date": self.to_date,
- "name": self.name,
- },
- as_dict=1,
+ shift = frappe.qb.DocType("Shift Request")
+ query = (
+ frappe.qb.from_(shift)
+ .select(shift.name, shift.shift_type)
+ .where((shift.employee == self.employee) & (shift.docstatus < 2) & (shift.name != self.name))
)
- for date_overlap in d:
- if date_overlap["name"]:
- self.throw_overlap_error(date_overlap)
+ if self.to_date:
+ query = query.where(
+ Criterion.any(
+ [
+ Criterion.any(
+ [
+ shift.to_date.isnull(),
+ ((self.from_date >= shift.from_date) & (self.from_date <= shift.to_date)),
+ ]
+ ),
+ Criterion.any(
+ [
+ ((self.to_date >= shift.from_date) & (self.to_date <= shift.to_date)),
+ shift.from_date.between(self.from_date, self.to_date),
+ ]
+ ),
+ ]
+ )
+ )
+ else:
+ query = query.where(
+ shift.to_date.isnull()
+ | ((self.from_date >= shift.from_date) & (self.from_date <= shift.to_date))
+ )
- def throw_overlap_error(self, d):
- msg = _("Employee {0} has already applied for {1} between {2} and {3} : ").format(
- self.employee, d["shift_type"], formatdate(d["from_date"]), formatdate(d["to_date"])
- ) + """ <b><a href="/app/Form/Shift Request/{0}">{0}</a></b>""".format(d["name"])
- frappe.throw(msg, OverlapError)
+ return query.run(as_dict=True)
+
+ def throw_overlap_error(self, shift_details):
+ shift_details = frappe._dict(shift_details)
+ msg = _(
+ "Employee {0} has already applied for Shift {1}: {2} that overlaps within this period"
+ ).format(
+ frappe.bold(self.employee),
+ frappe.bold(shift_details.shift_type),
+ get_link_to_form("Shift Request", shift_details.name),
+ )
+
+ frappe.throw(msg, title=_("Overlapping Shift Requests"), exc=OverlappingShiftRequestError)
diff --git a/erpnext/hr/doctype/shift_request/test_shift_request.py b/erpnext/hr/doctype/shift_request/test_shift_request.py
index b4f5177..c47418c 100644
--- a/erpnext/hr/doctype/shift_request/test_shift_request.py
+++ b/erpnext/hr/doctype/shift_request/test_shift_request.py
@@ -4,23 +4,24 @@
import unittest
import frappe
+from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_days, nowdate
from erpnext.hr.doctype.employee.test_employee import make_employee
+from erpnext.hr.doctype.shift_request.shift_request import OverlappingShiftRequestError
+from erpnext.hr.doctype.shift_type.test_shift_type import setup_shift_type
test_dependencies = ["Shift Type"]
-class TestShiftRequest(unittest.TestCase):
+class TestShiftRequest(FrappeTestCase):
def setUp(self):
- for doctype in ["Shift Request", "Shift Assignment"]:
- frappe.db.sql("delete from `tab{doctype}`".format(doctype=doctype))
-
- def tearDown(self):
- frappe.db.rollback()
+ for doctype in ["Shift Request", "Shift Assignment", "Shift Type"]:
+ frappe.db.delete(doctype)
def test_make_shift_request(self):
"Test creation/updation of Shift Assignment from Shift Request."
+ setup_shift_type(shift_type="Day Shift")
department = frappe.get_value("Employee", "_T-Employee-00001", "department")
set_shift_approver(department)
approver = frappe.db.sql(
@@ -48,6 +49,7 @@
self.assertEqual(shift_assignment_docstatus, 2)
def test_shift_request_approver_perms(self):
+ setup_shift_type(shift_type="Day Shift")
employee = frappe.get_doc("Employee", "_T-Employee-00001")
user = "test_approver_perm_emp@example.com"
make_employee(user, "_Test Company")
@@ -87,6 +89,145 @@
employee.shift_request_approver = ""
employee.save()
+ def test_overlap_for_request_without_to_date(self):
+ # shift should be Ongoing if Only from_date is present
+ user = "test_shift_request@example.com"
+ employee = make_employee(user, company="_Test Company", shift_request_approver=user)
+ setup_shift_type(shift_type="Day Shift")
+
+ shift_request = frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": "Day Shift",
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": nowdate(),
+ "approver": user,
+ "status": "Approved",
+ }
+ ).submit()
+
+ shift_request = frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": "Day Shift",
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": add_days(nowdate(), 2),
+ "approver": user,
+ "status": "Approved",
+ }
+ )
+
+ self.assertRaises(OverlappingShiftRequestError, shift_request.save)
+
+ def test_overlap_for_request_with_from_and_to_dates(self):
+ user = "test_shift_request@example.com"
+ employee = make_employee(user, company="_Test Company", shift_request_approver=user)
+ setup_shift_type(shift_type="Day Shift")
+
+ shift_request = frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": "Day Shift",
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": nowdate(),
+ "to_date": add_days(nowdate(), 30),
+ "approver": user,
+ "status": "Approved",
+ }
+ ).submit()
+
+ shift_request = frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": "Day Shift",
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": add_days(nowdate(), 10),
+ "to_date": add_days(nowdate(), 35),
+ "approver": user,
+ "status": "Approved",
+ }
+ )
+
+ self.assertRaises(OverlappingShiftRequestError, shift_request.save)
+
+ def test_overlapping_for_a_fixed_period_shift_and_ongoing_shift(self):
+ user = "test_shift_request@example.com"
+ employee = make_employee(user, company="_Test Company", shift_request_approver=user)
+
+ # shift setup for 8-12
+ shift_type = setup_shift_type(shift_type="Shift 1", start_time="08:00:00", end_time="12:00:00")
+ date = nowdate()
+
+ # shift with end date
+ frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": shift_type.name,
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": date,
+ "to_date": add_days(date, 30),
+ "approver": user,
+ "status": "Approved",
+ }
+ ).submit()
+
+ # shift setup for 11-15
+ shift_type = setup_shift_type(shift_type="Shift 2", start_time="11:00:00", end_time="15:00:00")
+ shift2 = frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": shift_type.name,
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": date,
+ "approver": user,
+ "status": "Approved",
+ }
+ )
+
+ self.assertRaises(OverlappingShiftRequestError, shift2.insert)
+
+ def test_allow_non_overlapping_shift_requests_for_same_day(self):
+ user = "test_shift_request@example.com"
+ employee = make_employee(user, company="_Test Company", shift_request_approver=user)
+
+ # shift setup for 8-12
+ shift_type = setup_shift_type(shift_type="Shift 1", start_time="08:00:00", end_time="12:00:00")
+ date = nowdate()
+
+ # shift with end date
+ frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": shift_type.name,
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": date,
+ "to_date": add_days(date, 30),
+ "approver": user,
+ "status": "Approved",
+ }
+ ).submit()
+
+ # shift setup for 13-15
+ shift_type = setup_shift_type(shift_type="Shift 2", start_time="13:00:00", end_time="15:00:00")
+ frappe.get_doc(
+ {
+ "doctype": "Shift Request",
+ "shift_type": shift_type.name,
+ "company": "_Test Company",
+ "employee": employee,
+ "from_date": date,
+ "approver": user,
+ "status": "Approved",
+ }
+ ).submit()
+
def set_shift_approver(department):
department_doc = frappe.get_doc("Department", department)