[REF]hr_holidays_timeoff_analysis: code refactoring

This commit is contained in:
2026-07-21 11:48:43 +02:00
committed by laetitiadacosta
parent 0dd1ff1dda
commit 4dcd312401
4 changed files with 235 additions and 60 deletions

View File

@@ -3,7 +3,7 @@
{ {
"name": "hr_holidays_timeoff_analysis", "name": "hr_holidays_timeoff_analysis",
"version": "18.0.1.0.0", "version": "18.0.2.0.0",
"author": "Elabore", "author": "Elabore",
"website": "https://git.elabore.coop/elabore/elabore-addons", "website": "https://git.elabore.coop/elabore/elabore-addons",
"maintainer": "Elabore", "maintainer": "Elabore",

View File

@@ -18,12 +18,12 @@ class TimeOffDay(models.Model):
hr_leave_type = fields.Many2one(related="hr_leave_id.holiday_status_id") hr_leave_type = fields.Many2one(related="hr_leave_id.holiday_status_id")
leave_duration_by_day = fields.Float() leave_duration_by_day = fields.Float()
def employee_is_scheduled_to_work_this_day(self, date, employee, leave): def employee_is_scheduled_to_work_this_day(self, date, employee, leave=None):
""" """
Check if the employee is scheduled to work on this day according to his Check if the employee is scheduled to work on this day according to his
calendar. calendar.
""" """
calendar = self.get_calendar(employee, date) calendar = self._get_calendar_for_leave(employee, date, leave)
if not calendar or not calendar.attendance_ids: if not calendar or not calendar.attendance_ids:
return False return False
day_of_week = str(date.weekday()) day_of_week = str(date.weekday())
@@ -41,7 +41,19 @@ class TimeOffDay(models.Model):
) )
return bool(attendances) return bool(attendances)
def get_calendar(self, employee, date, leave=None): def _get_calendar_for_leave(self, employee, date, leave=None):
"""
Return the calendar to use to know whether ``employee`` works on
``date`` for a given ``leave``.
The base implementation ignores ``leave`` and returns the employee's
regular working calendar. It is an extension point: the module
``hr_holidays_timeoff_analysis_l10n_fr_hr_holidays`` overrides it to
force the company calendar for French reference leaves.
"""
return self.get_calendar(employee, date)
def get_calendar(self, employee, date):
""" """
Get the working time calendar of the employee. Get the working time calendar of the employee.
""" """
@@ -125,12 +137,18 @@ class TimeOffDay(models.Model):
@api.model @api.model
def cron_manage_timeoff_days(self): def cron_manage_timeoff_days(self):
self.cron_create_timeoff_days() """Reconcile hr.leave.timeoff.day records day by day.
self.cron_delete_timeoff_days()
def cron_create_timeoff_days(self): On each day within the relevant date range, compute which timeoff.days
# Browse all validated leaves *should* exist (validated leaves covering the day, employee scheduled
leaves = self.env["hr.leave"].search( to work, not a public holiday) and reconcile against existing records:
- Stale records (leave dates changed, state changed, force-cancelled,
orphan, calendar/holiday change) are deleted.
- Missing records are created.
- Existing records with an outdated duration are updated in place.
"""
# Load all currently validated leaves that have dates and an employee.
valid_leaves = self.env["hr.leave"].search(
[ [
("state", "=", "validate"), ("state", "=", "validate"),
("request_date_from", "!=", False), ("request_date_from", "!=", False),
@@ -138,59 +156,104 @@ class TimeOffDay(models.Model):
("employee_id", "!=", False), ("employee_id", "!=", False),
] ]
) )
for leave in leaves: # If no validated leaves exist, all existing timeoff.days are stale/orphan.
current_date = leave.request_date_from if not valid_leaves:
employee = leave.employee_id self.search([]).unlink()
while current_date <= leave.request_date_to: return
# Determine the full date span to iterate over: from the earliest
# relevant date (oldest leave or oldest existing timeoff.day) to the
# latest relevant date (newest leave or newest existing timeoff.day).
#first, get the oldest and latest leave dates
leave_from_dates = valid_leaves.mapped("request_date_from")
leave_to_dates = valid_leaves.mapped("request_date_to")
start_date = min(leave_from_dates)
end_date = max(leave_to_dates)
#then, get oldest timeoff.day and latest timeoff.day
#and keep as start_date and end_date the min and max of both
oldest_td = self.search([], order="date asc", limit=1)
if oldest_td and oldest_td.date and oldest_td.date < start_date:
start_date = oldest_td.date
latest_td = self.search([], order="date desc", limit=1)
if latest_td and latest_td.date and latest_td.date > end_date:
end_date = latest_td.date
existing_tds = self.search(
[("date", ">=", start_date), ("date", "<=", end_date)]
)
# Build in-memory indexes for fast lookup per day.
# leaves_by_date: date -> list of validated leaves covering that day
leaves_by_date = {}
for leave in valid_leaves:
d = leave.request_date_from
while d <= leave.request_date_to:
if d not in leaves_by_date:
leaves_by_date[d] = []
leaves_by_date[d].append(leave)
d += timedelta(days=1)
# existing_by_date: date -> list of existing timeoff.day records
existing_by_date = {}
for td in existing_tds:
if td.date not in existing_by_date:
existing_by_date[td.date] = []
existing_by_date[td.date].append(td)
# Collect operations and apply them at the end to minimise DB round trips.
all_to_create = []
all_to_delete_ids = []
current_date = start_date
while current_date <= end_date:
covering_leaves = leaves_by_date.get(current_date, [])
# Compute the expected timeoff.days for this date.
# Key: (employee_id, leave_id) -> expected leave_duration_by_day
expected = {}
for leave in covering_leaves:
employee = leave.employee_id
if self.employee_is_scheduled_to_work_this_day( if self.employee_is_scheduled_to_work_this_day(
current_date, employee, leave current_date, employee, leave
) and not self._is_public_holiday_according_to_employe_tz( ) and not self._is_public_holiday_according_to_employe_tz(
current_date, employee current_date, employee
): ):
# The employee is scheduled to work this day expected[(employee.id, leave.id)] = (
# according his calendar and it's not a self.compute_leave_duration_by_day(leave)
# public holiday, )
# so create a timeoff day record if it does not already exist
if not self.search(
[
("date", "=", current_date),
("employee_id", "=", employee.id),
("hr_leave_id", "=", leave.id),
],
limit=1,
):
self.create(
{
"date": current_date,
"employee_id": employee.id,
"hr_leave_id": leave.id,
"leave_duration_by_day": (
self.compute_leave_duration_by_day(leave)
),
}
)
current_date += timedelta(days=1)
def cron_delete_timeoff_days(self): # Reconcile existing timeoff.days for this date.
# Browse all unvalidated or canceled leaves existing_today = existing_by_date.get(current_date, [])
leaves = self.env["hr.leave"].search( for td in existing_today:
[ key = (td.employee_id.id, td.hr_leave_id.id)
("state", "!=", "validate"), if key in expected:
("request_date_from", "!=", False), # Matched: keep the record, update duration if it changed.
("request_date_to", "!=", False), expected_duration = expected.pop(key)
("employee_id", "!=", False), if td.leave_duration_by_day != expected_duration:
] td.leave_duration_by_day = expected_duration
) else:
# Delete timeoff days for leaves that are no longer validated # No longer valid: leave dates changed, state changed,
for leave in leaves: # force-cancelled, orphan (hr_leave_id=False), calendar
self.search( # changed, or the day became a public holiday.
[ all_to_delete_ids.append(td.id)
("hr_leave_id", "=", leave.id),
] # Remaining entries in expected are missing timeoff.days -> create.
).unlink() for (employee_id, leave_id), duration in expected.items():
# Delete timeoff days that are not linked to any leave all_to_create.append(
self.search( {
[ "date": current_date,
("hr_leave_id", "=", False), "employee_id": employee_id,
] "hr_leave_id": leave_id,
).unlink() "leave_duration_by_day": duration,
}
)
current_date += timedelta(days=1)
# Apply batched operations.
if all_to_delete_ids:
self.browse(all_to_delete_ids).unlink()
if all_to_create:
self.create(all_to_create)

View File

@@ -1,2 +1,2 @@
id,name,model_id/id,group_id/id,perm_read,perm_write,perm_create,perm_unlink id,name,model_id/id,group_id/id,perm_read,perm_write,perm_create,perm_unlink
access_hr_leave_timeoff_day_manager,access_hr_holidays_timeoff_day_manager,model_hr_leave_timeoff_day,hr_holidays.group_hr_holidays_manager,1,0,0,0 access_hr_leave_timeoff_day_user,access_hr_holidays_timeoff_day_user,model_hr_leave_timeoff_day,hr_holidays.group_hr_holidays_user,1,0,0,0
1 id name model_id/id group_id/id perm_read perm_write perm_create perm_unlink
2 access_hr_leave_timeoff_day_manager access_hr_leave_timeoff_day_user access_hr_holidays_timeoff_day_manager access_hr_holidays_timeoff_day_user model_hr_leave_timeoff_day hr_holidays.group_hr_holidays_manager hr_holidays.group_hr_holidays_user 1 0 0 0

View File

@@ -717,3 +717,115 @@ class TestHrLeaveTimeoffDay(TransactionCase):
4, 4,
"leave_duration_by_day should be 4 day", "leave_duration_by_day should be 4 day",
) )
def test_leave_dates_modified(self):
leave = self.env["hr.leave"].create(
{
"employee_id": self.employee.id,
"request_date_from": Date.to_date("2025-09-01"), # Monday
"request_date_to": Date.to_date("2025-09-05"), # Friday
"holiday_status_id": self.time_off_type.id,
}
)
leave.state = "validate"
self.env["hr.leave.timeoff.day"].cron_manage_timeoff_days()
timeoff_days = self.env["hr.leave.timeoff.day"].search(
[
("employee_id", "=", self.employee.id),
("hr_leave_id", "=", leave.id),
]
)
self.assertEqual(
len(timeoff_days), 5, "Should have 5 timeoff days initially"
)
# Modify dates: Wed Sept 3 to Sun Sept 7
# Working days: Wed(3), Thu(4), Fri(5) — Mon(1), Tue(2) removed
# leave_skip_state_check: same escape hatch the core uses when
# back-office flows (calendar/contract change, batch generation)
# rewrite the dates of an already validated leave.
leave.with_context(leave_skip_state_check=True).write(
{
"request_date_from": Date.to_date("2025-09-03"),
"request_date_to": Date.to_date("2025-09-07"),
}
)
self.env["hr.leave.timeoff.day"].cron_manage_timeoff_days()
timeoff_days = self.env["hr.leave.timeoff.day"].search(
[
("employee_id", "=", self.employee.id),
("hr_leave_id", "=", leave.id),
]
)
self.assertEqual(
len(timeoff_days),
3,
"Should have 3 timeoff days after date modification",
)
updated_dates = timeoff_days.mapped("date")
self.assertIn(
Date.to_date("2025-09-03"),
updated_dates,
"Sept 3 should still be present",
)
self.assertIn(
Date.to_date("2025-09-04"),
updated_dates,
"Sept 4 should still be present",
)
self.assertIn(
Date.to_date("2025-09-05"),
updated_dates,
"Sept 5 should still be present",
)
self.assertNotIn(
Date.to_date("2025-09-01"),
updated_dates,
"Sept 1 should have been removed",
)
self.assertNotIn(
Date.to_date("2025-09-02"),
updated_dates,
"Sept 2 should have been removed",
)
def test_leave_duration_changed(self):
leave = self.env["hr.leave"].create(
{
"employee_id": self.employee.id,
"request_date_from": Date.to_date("2025-10-01"),
"request_date_to": Date.to_date("2025-10-01"),
"holiday_status_id": self.time_off_type.id,
}
)
leave.state = "validate"
self.env["hr.leave.timeoff.day"].cron_manage_timeoff_days()
timeoff_day = self.env["hr.leave.timeoff.day"].search(
[
("employee_id", "=", self.employee.id),
("hr_leave_id", "=", leave.id),
]
)
self.assertEqual(len(timeoff_day), 1)
self.assertEqual(
timeoff_day.leave_duration_by_day,
1.0,
"Should be 1.0 for a full day leave",
)
# Simulate a stale duration, then verify the cron corrects it
# in place (exercises the update-in-place reconciliation path).
timeoff_day.leave_duration_by_day = 0.5
self.env["hr.leave.timeoff.day"].cron_manage_timeoff_days()
timeoff_day = self.env["hr.leave.timeoff.day"].search(
[
("employee_id", "=", self.employee.id),
("hr_leave_id", "=", leave.id),
]
)
self.assertEqual(len(timeoff_day), 1)
self.assertEqual(
timeoff_day.leave_duration_by_day,
1.0,
"Should be corrected back to 1.0 for a full day leave",
)