from datetime import timedelta from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from rest_framework import status from .models import User, UserPrescription, ExerciseLog, RegistrationCode # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def make_therapist(email='therapist@example.com', password='pass1234'): return User.objects.create_user(email=email, password=password, is_therapist=True) def make_patient(email='patient@example.com', password='pass1234'): return User.objects.create_user(email=email, password=password) def get_tokens(client, email, password): resp = client.post('/api/auth/login/', {'email': email, 'password': password}, format='json') return resp.data['access'], resp.data['refresh'] def auth_client(client, token): client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}') return client # --------------------------------------------------------------------------- # RegistrationCode model # --------------------------------------------------------------------------- class RegistrationCodeModelTests(TestCase): def setUp(self): self.therapist = make_therapist() def test_valid_unused_fresh_code(self): code = RegistrationCode.objects.create(code='ABCD1234', therapist=self.therapist) self.assertTrue(code.is_valid()) def test_invalid_used_code(self): patient = make_patient() code = RegistrationCode.objects.create(code='USED0001', therapist=self.therapist, used_by=patient) self.assertFalse(code.is_valid()) def test_invalid_expired_code(self): code = RegistrationCode.objects.create(code='EXPD0001', therapist=self.therapist) # Back-date created_at beyond 28 hours RegistrationCode.objects.filter(pk=code.pk).update( created_at=timezone.now() - timedelta(hours=29) ) code.refresh_from_db() self.assertFalse(code.is_valid()) def test_valid_code_just_before_expiry(self): code = RegistrationCode.objects.create(code='ALMX0001', therapist=self.therapist) RegistrationCode.objects.filter(pk=code.pk).update( created_at=timezone.now() - timedelta(hours=27, minutes=59) ) code.refresh_from_db() self.assertTrue(code.is_valid()) def test_str_representation(self): code = RegistrationCode.objects.create(code='STR00001', therapist=self.therapist) self.assertIn('STR00001', str(code)) self.assertIn(self.therapist.email, str(code)) # --------------------------------------------------------------------------- # Health endpoint # --------------------------------------------------------------------------- class HealthTests(TestCase): def test_health_returns_ok(self): resp = self.client.get('/api/auth/health/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['status'], 'ok') # --------------------------------------------------------------------------- # Validate-code endpoint # --------------------------------------------------------------------------- class ValidateCodeTests(TestCase): def setUp(self): self.therapist = make_therapist() self.client = APIClient() def test_valid_code_returns_true(self): RegistrationCode.objects.create(code='VALID001', therapist=self.therapist) resp = self.client.post('/api/auth/validate-code/', {'code': 'VALID001'}, format='json') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertTrue(resp.data['valid']) def test_lowercase_code_normalised(self): RegistrationCode.objects.create(code='LOWER001', therapist=self.therapist) resp = self.client.post('/api/auth/validate-code/', {'code': 'lower001'}, format='json') self.assertTrue(resp.data['valid']) def test_unknown_code_returns_false(self): resp = self.client.post('/api/auth/validate-code/', {'code': 'XXXXXXXX'}, format='json') self.assertFalse(resp.data['valid']) def test_used_code_returns_false(self): patient = make_patient() RegistrationCode.objects.create(code='USEDX001', therapist=self.therapist, used_by=patient) resp = self.client.post('/api/auth/validate-code/', {'code': 'USEDX001'}, format='json') self.assertFalse(resp.data['valid']) # --------------------------------------------------------------------------- # Register endpoint # --------------------------------------------------------------------------- class RegisterTests(TestCase): def setUp(self): self.therapist = make_therapist() self.client = APIClient() RegistrationCode.objects.create(code='REGCODE1', therapist=self.therapist) def test_register_with_valid_code(self): resp = self.client.post('/api/auth/register/', { 'email': 'new@example.com', 'password': 'securepass', 'display_name': 'New User', 'invite_code': 'REGCODE1', }, format='json') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIn('access', resp.data) self.assertIn('refresh', resp.data) def test_register_marks_code_as_used(self): self.client.post('/api/auth/register/', { 'email': 'new2@example.com', 'password': 'securepass', 'display_name': 'New User', 'invite_code': 'REGCODE1', }, format='json') code = RegistrationCode.objects.get(code='REGCODE1') self.assertIsNotNone(code.used_by) def test_register_with_invalid_code_rejected(self): resp = self.client.post('/api/auth/register/', { 'email': 'bad@example.com', 'password': 'securepass', 'display_name': 'Bad', 'invite_code': 'BADCODE0', }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_register_reusing_code_rejected(self): # Use the code once self.client.post('/api/auth/register/', { 'email': 'first@example.com', 'password': 'securepass', 'display_name': 'First', 'invite_code': 'REGCODE1', }, format='json') # Try to use it again resp = self.client.post('/api/auth/register/', { 'email': 'second@example.com', 'password': 'securepass', 'display_name': 'Second', 'invite_code': 'REGCODE1', }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_register_duplicate_email_rejected(self): self.client.post('/api/auth/register/', { 'email': 'dup@example.com', 'password': 'securepass', 'display_name': 'Dup', 'invite_code': 'REGCODE1', }, format='json') RegistrationCode.objects.create(code='REGCODE2', therapist=self.therapist) resp = self.client.post('/api/auth/register/', { 'email': 'dup@example.com', 'password': 'securepass', 'display_name': 'Dup', 'invite_code': 'REGCODE2', }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) # --------------------------------------------------------------------------- # Login endpoint # --------------------------------------------------------------------------- class LoginTests(TestCase): def setUp(self): self.client = APIClient() self.patient = make_patient(email='login@example.com', password='correctpass') def test_login_valid_credentials(self): resp = self.client.post('/api/auth/login/', { 'email': 'login@example.com', 'password': 'correctpass' }, format='json') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIn('access', resp.data) def test_login_wrong_password(self): resp = self.client.post('/api/auth/login/', { 'email': 'login@example.com', 'password': 'wrongpass' }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_login_unknown_email(self): resp = self.client.post('/api/auth/login/', { 'email': 'nobody@example.com', 'password': 'anypass' }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) # --------------------------------------------------------------------------- # Me endpoint # --------------------------------------------------------------------------- class MeTests(TestCase): def setUp(self): self.client = APIClient() self.patient = make_patient(email='me@example.com', password='pass1234') token, _ = get_tokens(self.client, 'me@example.com', 'pass1234') auth_client(self.client, token) def test_me_returns_profile(self): resp = self.client.get('/api/auth/me/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['email'], 'me@example.com') self.assertFalse(resp.data['is_therapist']) def test_me_unauthenticated_rejected(self): self.client.credentials() resp = self.client.get('/api/auth/me/') self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) # --------------------------------------------------------------------------- # Patient list (therapist only) # --------------------------------------------------------------------------- class PatientListTests(TestCase): def setUp(self): self.client = APIClient() self.therapist = make_therapist(email='t@example.com') self.patient = make_patient(email='p@example.com') token, _ = get_tokens(self.client, 't@example.com', 'pass1234') auth_client(self.client, token) def test_therapist_can_list_patients(self): resp = self.client.get('/api/auth/patients/') self.assertEqual(resp.status_code, status.HTTP_200_OK) emails = [p['email'] for p in resp.data] self.assertIn('p@example.com', emails) self.assertNotIn('t@example.com', emails) def test_patient_cannot_list_patients(self): token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') auth_client(self.client, token) resp = self.client.get('/api/auth/patients/') self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_response_includes_prescription_fields(self): resp = self.client.get('/api/auth/patients/') patient_data = next(p for p in resp.data if p['email'] == 'p@example.com') self.assertIn('prescription', patient_data) self.assertIn('total_sessions', patient_data) self.assertIn('last_session_at', patient_data) # --------------------------------------------------------------------------- # Prescribe endpoint # --------------------------------------------------------------------------- class PrescribeTests(TestCase): def setUp(self): self.client = APIClient() self.therapist = make_therapist(email='t@example.com') self.patient = make_patient(email='p@example.com') token, _ = get_tokens(self.client, 't@example.com', 'pass1234') auth_client(self.client, token) def _prescribe(self, bpm=6.0, duration=300, locked=True): return self.client.put( f'/api/auth/patients/{self.patient.pk}/prescribe/', {'breathing_mode': 'custom', 'visualization_type': 'circle', 'bpm': bpm, 'duration_seconds': duration, 'locked': locked}, format='json' ) def test_therapist_can_prescribe(self): resp = self._prescribe(bpm=4.0, duration=360) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['bpm'], 4.0) self.assertEqual(resp.data['duration_seconds'], 360) def test_prescription_stored_in_db(self): self._prescribe(bpm=8.0, duration=180, locked=False) p = UserPrescription.objects.get(patient=self.patient) self.assertEqual(p.bpm, 8.0) self.assertEqual(p.duration_seconds, 180) self.assertFalse(p.locked) def test_update_existing_prescription(self): self._prescribe(bpm=6.0, duration=300) self._prescribe(bpm=10.0, duration=600) self.assertEqual(UserPrescription.objects.filter(patient=self.patient).count(), 1) p = UserPrescription.objects.get(patient=self.patient) self.assertEqual(p.bpm, 10.0) def test_patient_cannot_prescribe(self): token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') auth_client(self.client, token) resp = self._prescribe() self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_prescribe_nonexistent_patient_returns_404(self): resp = self.client.put( '/api/auth/patients/99999/prescribe/', {'breathing_mode': 'custom', 'visualization_type': 'circle', 'bpm': 6, 'duration_seconds': 300, 'locked': True}, format='json' ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_prescribe_locked_flag_respected(self): self._prescribe(locked=True) p = UserPrescription.objects.get(patient=self.patient) self.assertTrue(p.locked) # --------------------------------------------------------------------------- # My prescription endpoint # --------------------------------------------------------------------------- class MyPrescriptionTests(TestCase): def setUp(self): self.client = APIClient() self.therapist = make_therapist(email='t@example.com') self.patient = make_patient(email='p@example.com') def _patient_token(self): token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') return token def _therapist_token(self): token, _ = get_tokens(self.client, 't@example.com', 'pass1234') return token def test_no_prescription_returns_null(self): auth_client(self.client, self._patient_token()) resp = self.client.get('/api/auth/prescription/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIsNone(resp.data) def test_returns_prescription_after_assignment(self): # Therapist assigns auth_client(self.client, self._therapist_token()) self.client.put( f'/api/auth/patients/{self.patient.pk}/prescribe/', {'breathing_mode': 'custom', 'visualization_type': 'circle', 'bpm': 5.0, 'duration_seconds': 240, 'locked': True}, format='json' ) # Patient fetches auth_client(self.client, self._patient_token()) resp = self.client.get('/api/auth/prescription/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data['bpm'], 5.0) self.assertEqual(resp.data['duration_seconds'], 240) self.assertTrue(resp.data['locked']) # --------------------------------------------------------------------------- # Exercise log endpoint # --------------------------------------------------------------------------- class ExerciseLogTests(TestCase): def setUp(self): self.client = APIClient() self.patient = make_patient(email='p@example.com') token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') auth_client(self.client, token) def _log(self, duration=60, breaths=6): return self.client.post('/api/auth/log/', { 'breathing_mode': 'custom', 'visualization_type': 'circle', 'duration_seconds': duration, 'breath_count': breaths, }, format='json') def test_patient_can_log_exercise(self): resp = self._log(duration=90, breaths=8) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) self.assertEqual(resp.data['duration_seconds'], 90) self.assertEqual(resp.data['breath_count'], 8) def test_log_stored_in_db(self): self._log(duration=120, breaths=10) self.assertEqual(ExerciseLog.objects.filter(patient=self.patient).count(), 1) def test_multiple_logs_accumulate(self): self._log() self._log() self._log() self.assertEqual(ExerciseLog.objects.filter(patient=self.patient).count(), 3) def test_log_missing_required_field_rejected(self): resp = self.client.post('/api/auth/log/', { 'breathing_mode': 'custom', 'visualization_type': 'circle', # duration_seconds missing 'breath_count': 5, }, format='json') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_unauthenticated_log_rejected(self): self.client.credentials() resp = self._log() self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) # --------------------------------------------------------------------------- # Patient logs endpoint (therapist views patient history) # --------------------------------------------------------------------------- class PatientLogsTests(TestCase): def setUp(self): self.client = APIClient() self.therapist = make_therapist(email='t@example.com') self.patient = make_patient(email='p@example.com') # Create some logs for the patient for i in range(3): ExerciseLog.objects.create( patient=self.patient, breathing_mode='custom', visualization_type='circle', duration_seconds=60 + i * 10, breath_count=6 + i, ) token, _ = get_tokens(self.client, 't@example.com', 'pass1234') auth_client(self.client, token) def test_therapist_can_fetch_patient_logs(self): resp = self.client.get(f'/api/auth/patients/{self.patient.pk}/logs/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(len(resp.data), 3) def test_logs_contain_expected_fields(self): resp = self.client.get(f'/api/auth/patients/{self.patient.pk}/logs/') entry = resp.data[0] self.assertIn('id', entry) self.assertIn('duration_seconds', entry) self.assertIn('breath_count', entry) self.assertIn('completed_at', entry) def test_patient_cannot_fetch_other_patients_logs(self): token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') auth_client(self.client, token) resp = self.client.get(f'/api/auth/patients/{self.patient.pk}/logs/') self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_logs_for_unknown_patient_returns_404(self): resp = self.client.get('/api/auth/patients/99999/logs/') self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_logs_ordered_newest_first(self): resp = self.client.get(f'/api/auth/patients/{self.patient.pk}/logs/') durations = [e['duration_seconds'] for e in resp.data] self.assertEqual(durations, sorted(durations, reverse=True)) # --------------------------------------------------------------------------- # Invite codes endpoints # --------------------------------------------------------------------------- class InviteCodesTests(TestCase): def setUp(self): self.client = APIClient() self.therapist = make_therapist(email='t@example.com') token, _ = get_tokens(self.client, 't@example.com', 'pass1234') auth_client(self.client, token) def test_therapist_can_create_code(self): resp = self.client.post('/api/auth/codes/', {}, format='json') self.assertEqual(resp.status_code, status.HTTP_201_CREATED) self.assertIn('code', resp.data) self.assertEqual(len(resp.data['code']), 8) def test_therapist_can_list_codes(self): self.client.post('/api/auth/codes/', {}, format='json') self.client.post('/api/auth/codes/', {}, format='json') resp = self.client.get('/api/auth/codes/') self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(len(resp.data), 2) def test_code_status_is_pending_initially(self): resp = self.client.post('/api/auth/codes/', {}, format='json') self.assertEqual(resp.data['status'], 'pending') def test_therapist_can_revoke_pending_code(self): create_resp = self.client.post('/api/auth/codes/', {}, format='json') code_id = create_resp.data['id'] resp = self.client.delete(f'/api/auth/codes/{code_id}/') self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) self.assertFalse(RegistrationCode.objects.filter(pk=code_id).exists()) def test_cannot_revoke_used_code(self): patient = make_patient() code = RegistrationCode.objects.create( code='USEDC001', therapist=self.therapist, used_by=patient ) resp = self.client.delete(f'/api/auth/codes/{code.pk}/') self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) def test_patient_cannot_create_code(self): patient = make_patient(email='p@example.com') token, _ = get_tokens(self.client, 'p@example.com', 'pass1234') auth_client(self.client, token) resp = self.client.post('/api/auth/codes/', {}, format='json') self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_codes_only_visible_to_owning_therapist(self): # Second therapist make_therapist(email='t2@example.com') token2, _ = get_tokens(self.client, 't2@example.com', 'pass1234') # First therapist creates a code self.client.post('/api/auth/codes/', {}, format='json') # Second therapist lists — should see 0 auth_client(self.client, token2) resp = self.client.get('/api/auth/codes/') self.assertEqual(len(resp.data), 0)