from concurrent.futures import ThreadPoolExecutor from datetime import datetime, time from threading import Barrier from unittest.mock import patch from django.contrib.auth import get_user_model from django.db import connections from django.test import TransactionTestCase from django.utils import timezone from rest_framework.test import APIClient, APITestCase from .models import TimeEntry class TimeEntryAPITests(APITestCase): def setUp(self): self.user = get_user_model().objects.create_user(username='employee') self.other = get_user_model().objects.create_user(username='other') self.client.force_authenticate(self.user) self.payload = dict(start_date='2026-09-23', end_date='2026-09-23', start_time='08:00', end_time='12:00') def create_entry(self, **changes): response = self.client.post('/api/time/entries/create/', {**self.payload, **changes}, format='json') self.assertEqual(response.status_code, 201, response.data) return response.data def test_partial_updates_on_both_routes_and_date_changes(self): entry = self.create_entry() for suffix in ('', 'update/'): response = self.client.put(f'/api/time/entries/{entry["id"]}/{suffix}', {'start_time': '09:00'}, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertEqual(response.data['start_time'], '09:00:00') response = self.client.patch(f'/api/time/entries/{entry["id"]}/{suffix}', {'end_time': '07:00'}, format='json') self.assertEqual(response.status_code, 400) response = self.client.patch(f'/api/time/entries/{entry["id"]}/', {'start_date': '2026-09-24', 'end_date': '2026-09-24'}, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertEqual(response.data['start_date'], '2026-09-24') def test_owner_permissions(self): entry = self.create_entry() self.client.force_authenticate(self.other) self.assertEqual(self.client.get('/api/time/entries/').data, []) self.assertEqual(self.client.patch(f'/api/time/entries/{entry["id"]}/', {'end_time': '13:00'}, format='json').status_code, 404) self.assertEqual(self.client.delete(f'/api/time/entries/{entry["id"]}/').status_code, 404) self.assertEqual(self.client.get(f'/api/time/weekly-summary/{self.user.pk}/').status_code, 403) response = self.client.post('/api/time/entries/create/', {**self.payload, 'user_id': self.user.pk}, format='json') self.assertEqual(response.status_code, 403) def test_staff_can_create_for_employee(self): self.other.is_staff = True self.other.save() self.client.force_authenticate(self.other) entry = self.create_entry(user_id=self.user.pk) self.assertEqual(entry['user']['id'], self.user.pk) def test_overlap_rejected_but_adjacent_period_allowed(self): self.create_entry() for changes in ({}, {'start_time': '10:00', 'end_time': '14:00'}): response = self.client.post('/api/time/entries/create/', {**self.payload, **changes}, format='json') self.assertEqual(response.status_code, 400, response.data) self.create_entry(start_time='12:00', end_time='14:00') self.assertEqual(TimeEntry.objects.count(), 2) def test_group_update_delete_and_individual_route_protection(self): entry = self.create_entry(start_time='22:00', end_date='2026-09-25', end_time='06:00') url = f'/api/time/groups/{entry["group_id"]}/' self.assertEqual(TimeEntry.objects.count(), 3) self.assertEqual(self.client.patch(f'/api/time/entries/{entry["id"]}/', {'end_time': '23:00'}, format='json').status_code, 400) self.assertEqual(self.client.delete(f'/api/time/entries/{entry["id"]}/').status_code, 400) response = self.client.patch(url, {'start_time': '21:00', 'end_time': '07:00'}, format='json') self.assertEqual(response.status_code, 200, response.data) entries = list(TimeEntry.objects.order_by('start_date')) self.assertEqual(entries[0].start_time, time(21)) self.assertEqual(entries[-1].end_time, time(7)) self.client.force_authenticate(self.other) self.assertEqual(self.client.delete(url).status_code, 404) self.client.force_authenticate(self.user) self.assertEqual(self.client.delete(url).status_code, 204) self.assertFalse(TimeEntry.objects.exists()) def test_group_validation_does_not_partially_modify_rows(self): entry = self.create_entry(start_time='22:00', end_date='2026-09-24', end_time='06:00') self.create_entry(start_date='2026-09-24', end_date='2026-09-24', start_time='07:00', end_time='12:00') before = list(TimeEntry.objects.values()) response = self.client.patch(f'/api/time/groups/{entry["group_id"]}/', {'start_time': '21:00', 'end_time': '08:00'}, format='json') self.assertEqual(response.status_code, 400, response.data) self.assertEqual(list(TimeEntry.objects.values()), before) def test_group_write_failure_rolls_back(self): entry = self.create_entry(start_time='22:00', end_date='2026-09-24', end_time='06:00') before = list(TimeEntry.objects.values()) with patch('time_tracking.services.TimeEntry.objects.create', side_effect=RuntimeError('simulated failure')): with self.assertRaises(RuntimeError): self.client.patch(f'/api/time/groups/{entry["group_id"]}/', {'start_time': '21:00'}, format='json') self.assertEqual(list(TimeEntry.objects.values()), before) def test_single_entry_can_become_multiday(self): entry = self.create_entry() response = self.client.patch(f'/api/time/entries/{entry["id"]}/', {'end_date': '2026-09-24', 'end_time': '06:00'}, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertTrue(response.data['group_id']) self.assertEqual(TimeEntry.objects.count(), 2) def test_only_one_open_entry_allowed(self): self.create_entry(end_time=None) response = self.client.post('/api/time/entries/create/', {**self.payload, 'start_date': '2026-09-24', 'end_date': '2026-09-24', 'end_time': None}, format='json') self.assertEqual(response.status_code, 400) def test_scan_handles_previous_day_and_stale_resolution(self): now = timezone.make_aware(datetime(2026, 9, 24, 6)) entry = self.create_entry(start_time='22:00', end_time=None) codes = {'SECRET_POINTAGE_ENTER': 'enter-test', 'SECRET_POINTAGE_EXIT': 'exit-test'} with patch('time_tracking.views.config', side_effect=lambda key, default='': codes.get(key, default)), patch('time_tracking.views.timezone.localtime', return_value=now): response = self.client.post('/api/time/entries/scan/', {'code': 'enter-test'}, format='json') self.assertEqual(response.status_code, 409, response.data) self.assertEqual(TimeEntry.objects.count(), 1) conflict = response.data response = self.client.post('/api/time/entries/scan/', { 'code': 'enter-test', 'resolution': conflict['resolution'], 'time': '06:00', 'scan_date': conflict['scan_date'], 'entry_id': conflict['entry_id'], }, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertEqual(TimeEntry.objects.count(), 2) self.assertFalse(TimeEntry.objects.filter(end_time__isnull=True).exists()) def test_scan_missing_entry_requires_original_scan_and_rejects_replay(self): now = timezone.make_aware(datetime(2026, 9, 23, 12)) codes = {'SECRET_POINTAGE_ENTER': 'enter-test', 'SECRET_POINTAGE_EXIT': 'exit-test'} with patch('time_tracking.views.config', side_effect=lambda key, default='': codes.get(key, default)), patch('time_tracking.views.timezone.localtime', return_value=now): conflict = self.client.post('/api/time/entries/scan/', {'code': 'exit-test'}, format='json') self.assertEqual(conflict.status_code, 409) data = dict(code='exit-test', resolution='complete_missing_entry', time='08:00', scan_time=conflict.data['scan_time'], scan_date=conflict.data['scan_date']) self.assertEqual(self.client.post('/api/time/entries/scan/', data, format='json').status_code, 200) self.assertEqual(self.client.post('/api/time/entries/scan/', data, format='json').status_code, 400) self.assertEqual(TimeEntry.objects.count(), 1) data['scan_date'] = '2026-09-22' self.assertEqual(self.client.post('/api/time/entries/scan/', data, format='json').status_code, 400) def test_unconfigured_scan_fails_closed(self): with patch('time_tracking.views.config', return_value=''): self.assertEqual(self.client.post('/api/time/entries/scan/', {}, format='json').status_code, 503) self.assertFalse(TimeEntry.objects.exists()) class ConcurrentTimeEntryTests(TransactionTestCase): def test_concurrent_first_entries_are_serialized_per_user(self): user = get_user_model().objects.create_user(username='concurrent') ready = Barrier(2) def create(start_time): try: client = APIClient() client.force_authenticate(user) ready.wait(timeout=10) return client.post('/api/time/entries/create/', { 'start_date': '2026-09-23', 'start_time': start_time, }, format='json').status_code finally: connections.close_all() with ThreadPoolExecutor(max_workers=2) as executor: results = list(executor.map(create, ['08:00', '08:01'])) self.assertEqual(sorted(results), [201, 400]) self.assertEqual(TimeEntry.objects.filter(end_time__isnull=True).count(), 1)