170 lines
9.9 KiB
Python
170 lines
9.9 KiB
Python
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)
|