302 lines
18 KiB
Python
302 lines
18 KiB
Python
from datetime import date
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from rest_framework.test import APITestCase
|
|
|
|
from time_tracking.models import TimeEntry
|
|
from .models import PlanningState, WorkSchedule, ScheduleException, DayChangeRequest
|
|
from .services import effective_period
|
|
|
|
|
|
class PlanningAPITests(APITestCase):
|
|
def setUp(self):
|
|
self.admin = get_user_model().objects.create_user(username='admin', is_staff=True)
|
|
self.alice = get_user_model().objects.create_user(username='alice')
|
|
self.bob = get_user_model().objects.create_user(username='bob')
|
|
self.stranger = get_user_model().objects.create_user(username='other')
|
|
self.client.force_authenticate(self.admin)
|
|
|
|
def slot(self, start='08:00', end='12:00', weekday=None, offset=0):
|
|
result = dict(start_time=start, end_time=end, end_day_offset=offset)
|
|
if weekday is not None:
|
|
result['weekday'] = weekday
|
|
return result
|
|
|
|
def schedule(self, user, slots=None, start='2030-01-07'):
|
|
state = PlanningState.objects.filter(user=user).first()
|
|
return self.client.post('/api/planning/schedules/', dict(user_id=user.pk, valid_from=start,
|
|
expected_revision=state.revision if state else 0, slots=slots if slots is not None else [self.slot(weekday=0)]), format='json')
|
|
|
|
def exception(self, **values):
|
|
state = PlanningState.objects.filter(user=self.alice).first()
|
|
data = dict(user_id=self.alice.pk, expected_revision=state.revision if state else 0,
|
|
label='Exception', kind='work', start_date='2030-01-07', end_date=None,
|
|
recurring=True, weekdays=[0], every_weeks=2, slots=[self.slot('10:00', '14:00')])
|
|
data.update(values)
|
|
return self.client.post('/api/planning/exceptions/', data, format='json')
|
|
|
|
def request(self, **values):
|
|
self.client.force_authenticate(self.alice)
|
|
data = dict(kind='leave', start_date='2030-01-07', end_date='2030-01-07', reason='Test', proposed_slots=[])
|
|
data.update(values)
|
|
return self.client.post('/api/planning/requests/', data, format='json')
|
|
|
|
def decide(self, row_id, action='approve', target_date=None):
|
|
self.client.force_authenticate(self.admin)
|
|
payload = {'action': action, 'response': 'Réponse admin'}
|
|
if target_date:
|
|
payload['target_date'] = target_date
|
|
return self.client.post(f'/api/planning/requests/{row_id}/decision/', payload, format='json')
|
|
|
|
def test_permissions_and_minimal_people_directory(self):
|
|
self.client.force_authenticate(self.alice)
|
|
self.assertEqual(self.client.get('/api/planning/config/').status_code, 403)
|
|
self.assertEqual(self.client.get('/api/planning/effective/', {'user_id': self.bob.pk}).status_code, 403)
|
|
self.assertEqual(self.schedule(self.alice).status_code, 403)
|
|
directory = self.client.get('/api/planning/people/').data
|
|
self.assertNotIn('email', directory[0])
|
|
self.assertNotIn('is_staff', directory[0])
|
|
|
|
def test_new_effective_date_preserves_history_and_inserts_between_versions(self):
|
|
self.assertEqual(self.schedule(self.alice).status_code, 201)
|
|
self.assertEqual(self.schedule(self.alice, [self.slot('10:00', '14:00', 0)], '2030-02-04').status_code, 201)
|
|
self.assertEqual(self.schedule(self.alice, [self.slot('09:00', '13:00', 0)], '2030-01-21').status_code, 201)
|
|
for day, expected in [('2030-01-14', '08:00'), ('2030-01-28', '09:00'), ('2030-02-11', '10:00')]:
|
|
result = effective_period(self.alice.pk, date.fromisoformat(day), date.fromisoformat(day))
|
|
self.assertEqual(result['segments'][0]['start_time'], expected)
|
|
self.assertEqual(WorkSchedule.objects.get(user=self.alice, valid_from='2030-01-07').valid_to.isoformat(), '2030-01-20')
|
|
|
|
def test_revision_rejects_lost_admin_update(self):
|
|
self.schedule(self.alice)
|
|
result = self.client.post('/api/planning/schedules/', dict(user_id=self.alice.pk, valid_from='2030-01-07', expected_revision=0, slots=[]), format='json')
|
|
self.assertEqual(result.status_code, 409)
|
|
self.assertEqual(WorkSchedule.objects.get(user=self.alice).slots.count(), 1)
|
|
|
|
def test_recurring_exceptions_and_priority(self):
|
|
self.schedule(self.alice)
|
|
self.assertEqual(self.exception().status_code, 201)
|
|
self.assertEqual(self.exception(recurring=False, start_date='2030-01-21', end_date='2030-01-21', kind='absence', slots=[]).status_code, 201)
|
|
for day, expected in [('2030-01-07', '10:00'), ('2030-01-14', '08:00'), ('2030-01-28', '08:00'), ('2030-02-04', '10:00')]:
|
|
data = effective_period(self.alice.pk, date.fromisoformat(day), date.fromisoformat(day))
|
|
self.assertEqual(data['segments'][0]['start_time'], expected)
|
|
data = effective_period(self.alice.pk, date(2030, 1, 21), date(2030, 1, 21))
|
|
self.assertEqual(data['days'][0]['kind'], 'absence')
|
|
self.assertEqual(data['segments'], [])
|
|
|
|
def test_overnight_previous_day_and_week_boundary(self):
|
|
self.assertEqual(self.schedule(self.alice, [self.slot('22:00', '06:00', 6, 1)]).status_code, 201)
|
|
data = effective_period(self.alice.pk, date(2030, 1, 14), date(2030, 1, 14))
|
|
self.assertEqual([(s['start_time'], s['end_time']) for s in data['segments']], [('00:00', '06:00')])
|
|
result = self.schedule(self.alice, [self.slot('22:00', '06:00', 6, 1), self.slot('05:00', '09:00', 0)])
|
|
self.assertEqual(result.status_code, 400)
|
|
self.assertEqual(WorkSchedule.objects.get(user=self.alice).slots.count(), 1)
|
|
|
|
def test_far_future_recurring_conflict_rolls_back(self):
|
|
self.schedule(self.alice, [self.slot('22:00', '06:00', 6, 1)])
|
|
result = self.exception(start_date='2037-01-01', weekdays=[0], every_weeks=3, slots=[self.slot('04:00', '08:00')])
|
|
self.assertEqual(result.status_code, 400)
|
|
self.assertFalse(ScheduleException.objects.exists())
|
|
|
|
def test_leave_approval_preserves_real_time_entries_and_prevents_replay(self):
|
|
self.schedule(self.alice)
|
|
actual = TimeEntry.objects.create(user=self.alice, start_date='2030-01-07', end_date='2030-01-07', start_time='08:03', end_time='12:01')
|
|
row = self.request()
|
|
self.assertEqual(row.status_code, 201, row.data)
|
|
self.assertEqual(self.decide(row.data['id']).status_code, 200)
|
|
actual.refresh_from_db()
|
|
self.assertEqual(str(actual.start_time), '08:03:00')
|
|
self.assertEqual(self.decide(row.data['id']).status_code, 409)
|
|
self.assertEqual(ScheduleException.objects.count(), 1)
|
|
self.assertEqual(effective_period(self.alice.pk, date(2030, 1, 7), date(2030, 1, 7))['days'][0]['kind'], 'absence')
|
|
|
|
def test_stale_request_is_not_applied(self):
|
|
self.schedule(self.alice)
|
|
row = self.request()
|
|
self.client.force_authenticate(self.admin)
|
|
self.schedule(self.alice, [self.slot('09:00', '13:00', 0)])
|
|
self.assertEqual(self.decide(row.data['id']).status_code, 409)
|
|
self.assertFalse(ScheduleException.objects.exists())
|
|
self.assertEqual(DayChangeRequest.objects.get().status, 'pending')
|
|
|
|
def test_move_requires_same_total_and_moves_work_to_another_day(self):
|
|
self.schedule(self.alice)
|
|
row = self.request(kind='move', target_date='2030-01-08', proposed_slots=[self.slot('13:00', '17:00')])
|
|
self.assertEqual(row.status_code, 201, row.data)
|
|
self.assertEqual(self.decide(row.data['id']).status_code, 200)
|
|
days = effective_period(self.alice.pk, date(2030, 1, 7), date(2030, 1, 8))['days']
|
|
self.assertEqual([d['kind'] for d in days], ['rest', 'work'])
|
|
self.assertEqual(days[1]['slots'][0]['start_time'], '13:00')
|
|
|
|
self.client.force_authenticate(self.alice)
|
|
wrong_total = self.client.post('/api/planning/requests/', dict(
|
|
kind='move', start_date='2030-01-14', end_date='2030-01-14', target_date='2030-01-15',
|
|
proposed_slots=[self.slot('13:00', '16:00')],
|
|
), format='json')
|
|
self.assertEqual(wrong_total.status_code, 400)
|
|
|
|
def test_two_employee_swap_and_request_visibility(self):
|
|
self.schedule(self.alice)
|
|
self.schedule(self.bob, [self.slot('13:00', '18:00', 0)])
|
|
row = self.request(kind='swap', target_user_id=self.bob.pk)
|
|
self.assertEqual(row.status_code, 201, row.data)
|
|
self.assertIsNone(row.data['target_date'])
|
|
self.assertEqual(row.data['after'], [])
|
|
self.client.force_authenticate(self.bob)
|
|
self.assertEqual(self.client.get('/api/planning/requests/').data['count'], 1)
|
|
self.assertEqual(self.client.post(f"/api/planning/requests/{row.data['id']}/decision/", {'action': 'cancel'}, format='json').status_code, 403)
|
|
self.client.force_authenticate(self.stranger)
|
|
self.assertEqual(self.client.get('/api/planning/requests/').data['count'], 0)
|
|
# Le responsable choisit une autre journée du collègue au moment de valider.
|
|
self.assertEqual(self.decide(row.data['id'], target_date='2030-01-14').status_code, 200)
|
|
self.assertEqual(effective_period(self.alice.pk, date(2030, 1, 7), date(2030, 1, 7))['segments'][0]['start_time'], '13:00')
|
|
self.assertEqual(effective_period(self.bob.pk, date(2030, 1, 14), date(2030, 1, 14))['segments'][0]['start_time'], '08:00')
|
|
|
|
def test_swap_target_is_chosen_by_admin_and_move_rejects_overlap(self):
|
|
self.schedule(self.alice, [self.slot('08:00', '12:00', 0), self.slot('13:00', '18:00', 1)])
|
|
self.schedule(self.bob, [self.slot('13:00', '18:00', 0)])
|
|
row = self.request(kind='swap', target_user_id=self.bob.pk)
|
|
self.assertEqual(row.status_code, 201, row.data)
|
|
# Avant le début du planning de Bob : journée non définie, donc impossible à échanger.
|
|
self.assertEqual(self.decide(row.data['id'], target_date='2030-01-06').status_code, 400)
|
|
self.assertEqual(DayChangeRequest.objects.get(pk=row.data['id']).status, 'pending')
|
|
|
|
self.client.force_authenticate(self.alice)
|
|
overlap = self.client.post('/api/planning/requests/', dict(
|
|
kind='move', start_date='2030-01-08', end_date='2030-01-08', target_date='2030-01-09',
|
|
proposed_slots=[self.slot('14:00', '18:00')],
|
|
), format='json')
|
|
self.assertEqual(overlap.status_code, 400)
|
|
|
|
def test_duplicate_cancel_and_reject_do_not_modify_schedule(self):
|
|
self.schedule(self.alice)
|
|
row = self.request()
|
|
self.assertEqual(self.request().status_code, 409)
|
|
self.assertEqual(self.client.post(f"/api/planning/requests/{row.data['id']}/decision/", {'action': 'approve'}, format='json').status_code, 403)
|
|
self.assertEqual(self.client.post(f"/api/planning/requests/{row.data['id']}/decision/", {'action': 'cancel'}, format='json').status_code, 200)
|
|
row = self.request()
|
|
self.assertEqual(self.decide(row.data['id'], 'reject').status_code, 200)
|
|
self.assertFalse(ScheduleException.objects.exists())
|
|
|
|
def test_preview_changes_are_rechecked_when_request_is_submitted(self):
|
|
self.schedule(self.alice)
|
|
self.client.force_authenticate(self.alice)
|
|
payload = dict(kind='leave', start_date='2030-01-07', end_date='2030-01-07')
|
|
preview = self.client.post('/api/planning/requests/preview/', payload, format='json')
|
|
self.assertEqual(preview.status_code, 200)
|
|
self.client.force_authenticate(self.admin)
|
|
self.schedule(self.alice, [self.slot('10:00', '14:00', 0)])
|
|
self.client.force_authenticate(self.alice)
|
|
payload['expected_before'] = preview.data['before']
|
|
response = self.client.post('/api/planning/requests/', payload, format='json')
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertFalse(DayChangeRequest.objects.exists())
|
|
|
|
def test_weekly_slots_validate_order_and_dates(self):
|
|
response = self.schedule(self.alice, [self.slot('12:00', '08:00', 0)])
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertFalse(WorkSchedule.objects.exists())
|
|
self.assertEqual(self.exception(weekdays=[], recurring=True).status_code, 400)
|
|
|
|
def test_daily_and_weekly_hour_targets_are_exposed_by_effective_api(self):
|
|
daily = self.client.post('/api/planning/schedules/', dict(
|
|
user_id=self.alice.pk,
|
|
valid_from='2030-01-07',
|
|
cycle_weeks=1,
|
|
mode='daily_hours',
|
|
day_targets=[
|
|
{'week_index': 0, 'weekday': 0, 'minutes': 450},
|
|
{'week_index': 0, 'weekday': 1, 'minutes': 360},
|
|
],
|
|
week_targets=[],
|
|
slots=[],
|
|
expected_revision=0,
|
|
), format='json')
|
|
self.assertEqual(daily.status_code, 201, daily.data)
|
|
result = self.client.get('/api/planning/effective/', dict(user_id=self.alice.pk, start='2030-01-07', end='2030-01-08')).data
|
|
self.assertEqual(result['days'][0]['schedule_mode'], 'daily_hours')
|
|
self.assertEqual(result['days'][0]['target_minutes'], 450)
|
|
self.assertEqual(result['days'][0]['week_target_minutes'], 810)
|
|
self.assertEqual(result['days'][0]['slots'], [])
|
|
self.assertEqual(result['days'][1]['target_minutes'], 360)
|
|
|
|
revision = PlanningState.objects.get(user=self.alice).revision
|
|
weekly = self.client.post('/api/planning/schedules/', dict(
|
|
user_id=self.alice.pk,
|
|
valid_from='2030-01-14',
|
|
cycle_weeks=1,
|
|
mode='weekly_hours',
|
|
day_targets=[],
|
|
week_targets=[{'week_index': 0, 'minutes': 2100}],
|
|
slots=[],
|
|
expected_revision=revision,
|
|
), format='json')
|
|
self.assertEqual(weekly.status_code, 201, weekly.data)
|
|
result = self.client.get('/api/planning/effective/', dict(user_id=self.alice.pk, start='2030-01-14', end='2030-01-20')).data
|
|
self.assertTrue(all(day['schedule_mode'] == 'weekly_hours' for day in result['days']))
|
|
self.assertTrue(all(day['week_target_minutes'] == 2100 for day in result['days']))
|
|
self.assertEqual(result['segments'], [])
|
|
|
|
def rotating_schedule(self, weeks, slots, start='2030-01-07'):
|
|
state = PlanningState.objects.filter(user=self.alice).first()
|
|
return self.client.post('/api/planning/schedules/', dict(user_id=self.alice.pk, valid_from=start,
|
|
cycle_weeks=weeks, expected_revision=state.revision if state else 0, slots=slots), format='json')
|
|
|
|
def cycle_slot(self, week, weekday=0, start='08:00', end='12:00', offset=0):
|
|
return {**self.slot(start, end, weekday, offset), 'week_index': week}
|
|
|
|
def test_two_weeks_alternate_in_effective_api(self):
|
|
result = self.rotating_schedule(2, [self.cycle_slot(0), self.cycle_slot(1, start='13:00', end='17:00')])
|
|
self.assertEqual(result.status_code, 201, result.data)
|
|
self.assertEqual(result.data['cycle_weeks'], 2)
|
|
for day, hour in [('2030-01-07', '08:00'), ('2030-01-14', '13:00'), ('2030-01-21', '08:00')]:
|
|
response = self.client.get('/api/planning/effective/', dict(user_id=self.alice.pk, start=day, end=day))
|
|
self.assertEqual(response.data['segments'][0]['start_time'], hour)
|
|
|
|
def test_three_week_cycle_includes_empty_rest_week(self):
|
|
self.assertEqual(self.rotating_schedule(3, [self.cycle_slot(0), self.cycle_slot(2, start='10:00', end='14:00')]).status_code, 201)
|
|
days = effective_period(self.alice.pk, date(2030, 1, 7), date(2030, 1, 28))['days']
|
|
self.assertEqual([days[index]['kind'] for index in [0, 7, 14, 21]], ['work', 'rest', 'work', 'work'])
|
|
self.assertEqual(days[14]['slots'][0]['start_time'], '10:00')
|
|
|
|
def test_midweek_effective_date_anchors_a_to_containing_monday(self):
|
|
self.rotating_schedule(2, [self.cycle_slot(0, 2), self.cycle_slot(1, 0, '13:00', '17:00')], start='2030-01-09')
|
|
days = effective_period(self.alice.pk, date(2030, 1, 8), date(2030, 1, 14))['days']
|
|
self.assertEqual(days[0]['kind'], 'undefined')
|
|
self.assertEqual(days[1]['slots'][0]['start_time'], '08:00')
|
|
self.assertEqual(days[-1]['slots'][0]['start_time'], '13:00')
|
|
|
|
def test_cycle_change_preserves_previous_period(self):
|
|
self.rotating_schedule(2, [self.cycle_slot(0), self.cycle_slot(1, start='13:00', end='17:00')])
|
|
self.rotating_schedule(3, [self.cycle_slot(0, start='09:00', end='12:00')], start='2030-01-21')
|
|
self.assertEqual(effective_period(self.alice.pk, date(2030, 1, 14), date(2030, 1, 14))['segments'][0]['start_time'], '13:00')
|
|
self.assertEqual(effective_period(self.alice.pk, date(2030, 1, 21), date(2030, 1, 21))['segments'][0]['start_time'], '09:00')
|
|
|
|
def test_cycle_wrap_rejects_overnight_collision_and_keeps_previous_version(self):
|
|
slots = [self.cycle_slot(2, 6, '22:00', '06:00', 1), self.cycle_slot(0, 0, '07:00', '12:00')]
|
|
self.assertEqual(self.rotating_schedule(3, slots).status_code, 201)
|
|
data = effective_period(self.alice.pk, date(2030, 1, 28), date(2030, 1, 28))
|
|
self.assertEqual([(s['start_time'], s['end_time']) for s in data['segments']], [('00:00', '06:00'), ('07:00', '12:00')])
|
|
slots[1]['start_time'] = '05:00'
|
|
self.assertEqual(self.rotating_schedule(3, slots).status_code, 400)
|
|
self.assertEqual(WorkSchedule.objects.get().slots.get(week_index=0).start_time.hour, 7)
|
|
|
|
def test_cycle_and_exception_conflict_after_52_weeks_is_detected(self):
|
|
self.assertEqual(self.rotating_schedule(13, [self.cycle_slot(12, 6, '22:00', '06:00', 1)]).status_code, 201)
|
|
result = self.exception(every_weeks=4, weekdays=[0], slots=[self.slot('04:00', '08:00')])
|
|
self.assertEqual(result.status_code, 400, result.data)
|
|
self.assertFalse(ScheduleException.objects.exists())
|
|
|
|
def test_cycle_validation_rejects_nonexistent_weeks(self):
|
|
self.assertEqual(self.rotating_schedule(2, [self.cycle_slot(2)]).status_code, 400)
|
|
self.assertEqual(self.rotating_schedule(0, []).status_code, 400)
|
|
self.assertEqual(self.rotating_schedule(53, []).status_code, 400)
|
|
self.assertFalse(WorkSchedule.objects.exists())
|
|
|
|
def test_approved_leave_overrides_only_one_day_in_cycle(self):
|
|
self.rotating_schedule(2, [self.cycle_slot(0), self.cycle_slot(1, start='13:00', end='17:00')])
|
|
row = self.request(start_date='2030-01-14', end_date='2030-01-14')
|
|
self.assertEqual(self.decide(row.data['id']).status_code, 200)
|
|
days = effective_period(self.alice.pk, date(2030, 1, 14), date(2030, 1, 28))['days']
|
|
self.assertEqual(days[0]['kind'], 'absence')
|
|
self.assertEqual(days[-1]['slots'][0]['start_time'], '13:00')
|