from django.contrib.auth import get_user_model from django.test import Client, TestCase from django.urls import reverse from rest_framework.authtoken.models import Token from rest_framework.test import APIClient from jobs.models import Job, JobApplication from messaging.models import Conversation, Message from messaging.services import ( ConversationClosedError, close_conversation, get_or_create_conversation, reopen_conversation, send_message, start_conversation_for_application, ) from seekers.models import SeekerProfile User = get_user_model() class MessagingTestCase(TestCase): def setUp(self): self.employer = User.objects.create_user( username='employer1', email='employer@example.com', password='pass12345', role='company', ) self.seeker_user = User.objects.create_user( username='seeker1', email='seeker@example.com', password='pass12345', role='seeker', ) SeekerProfile.objects.create( user=self.seeker_user, full_name='Jane Seeker', headline='Developer', location='Nairobi', ) self.job = Job.objects.create( title='Software Engineer', slug='software-engineer-test', company_name='Acme Ltd', company_slug='acme-ltd', company_initials='AL', location='Lusaka, Lusaka, Zambia', job_type='remote', description='Build things', responsibilities='Code', requirements='Python', posted_by=self.employer, is_active=True, ) self.application = JobApplication.objects.create( job=self.job, applicant=self.seeker_user, status='pending', ) self.client = Client() def test_start_conversation_from_application(self): conversation = start_conversation_for_application(self.employer, self.application.pk) self.assertIsNotNone(conversation) self.assertEqual(conversation.application_id, self.application.pk) self.assertTrue(conversation.is_open) def test_company_and_seeker_can_exchange_messages(self): conversation = get_or_create_conversation(self.application) msg1 = send_message(conversation, self.employer, 'Hello Jane') msg2 = send_message(conversation, self.seeker_user, 'Hi there') self.assertEqual(conversation.messages.count(), 2) self.assertEqual(msg1.sender, self.employer) self.assertEqual(msg2.sender, self.seeker_user) def test_company_can_close_and_reopen_conversation(self): conversation = get_or_create_conversation(self.application) self.assertTrue(close_conversation(conversation, self.employer)) conversation.refresh_from_db() self.assertTrue(conversation.is_closed) self.assertIsNotNone(conversation.closed_at) self.assertEqual(conversation.closed_by, self.employer) with self.assertRaises(ConversationClosedError): send_message(conversation, self.employer, 'Should fail') self.assertTrue(reopen_conversation(conversation, self.employer)) conversation.refresh_from_db() self.assertTrue(conversation.is_open) send_message(conversation, self.employer, 'Back open') def test_seeker_cannot_close_conversation(self): conversation = get_or_create_conversation(self.application) self.assertFalse(close_conversation(conversation, self.seeker_user)) conversation.refresh_from_db() self.assertTrue(conversation.is_open) def test_inbox_start_and_send_flow(self): self.client.login(username='employer1', password='pass12345') start_url = reverse('messaging:start', kwargs={'application_id': self.application.pk}) response = self.client.get(start_url) self.assertEqual(response.status_code, 302) conversation = Conversation.objects.get(application=self.application) send_url = reverse('messaging:send', kwargs={'pk': conversation.pk}) response = self.client.post(send_url, {'body': 'We would like to interview you.'}) self.assertEqual(response.status_code, 302) self.assertEqual(Message.objects.filter(conversation=conversation).count(), 1) close_url = reverse('messaging:close', kwargs={'pk': conversation.pk}) response = self.client.post(close_url) self.assertEqual(response.status_code, 302) conversation.refresh_from_db() self.assertTrue(conversation.is_closed) response = self.client.post(send_url, {'body': 'Another message'}) self.assertEqual(response.status_code, 302) self.assertEqual(Message.objects.filter(conversation=conversation).count(), 1) def test_seeker_can_read_closed_conversation_inbox(self): conversation = get_or_create_conversation(self.application) send_message(conversation, self.employer, 'Thanks for applying') close_conversation(conversation, self.employer) self.client.login(username='seeker1', password='pass12345') inbox_url = reverse('messaging:inbox_conversation', kwargs={'pk': conversation.pk}) response = self.client.get(inbox_url) self.assertEqual(response.status_code, 200) self.assertContains(response, 'This conversation is closed') self.assertContains(response, 'Messaging is disabled') def test_seeker_can_start_and_reply(self): self.client.login(username='seeker1', password='pass12345') start_url = reverse('messaging:start', kwargs={'application_id': self.application.pk}) response = self.client.get(start_url) self.assertEqual(response.status_code, 302) conversation = Conversation.objects.get(application=self.application) send_url = reverse('messaging:send', kwargs={'pk': conversation.pk}) response = self.client.post(send_url, {'body': 'Thank you for considering my application.'}) self.assertEqual(response.status_code, 302) self.assertEqual(Message.objects.filter(conversation=conversation, sender=self.seeker_user).count(), 1) def test_reopen_via_web_allows_messaging_again(self): conversation = get_or_create_conversation(self.application) close_conversation(conversation, self.employer) self.client.login(username='employer1', password='pass12345') reopen_url = reverse('messaging:reopen', kwargs={'pk': conversation.pk}) response = self.client.post(reopen_url) self.assertEqual(response.status_code, 302) conversation.refresh_from_db() self.assertTrue(conversation.is_open) send_url = reverse('messaging:send', kwargs={'pk': conversation.pk}) response = self.client.post(send_url, {'body': 'Follow-up question'}) self.assertEqual(response.status_code, 302) self.assertEqual(Message.objects.filter(conversation=conversation).count(), 1) def test_inbox_page_renders_for_both_roles(self): conversation = get_or_create_conversation(self.application) send_message(conversation, self.employer, 'Hello') self.client.login(username='employer1', password='pass12345') response = self.client.get(reverse('messaging:inbox')) self.assertEqual(response.status_code, 200) self.assertContains(response, 'Close chat') self.assertContains(response, 'Hello') self.client.login(username='seeker1', password='pass12345') response = self.client.get(reverse('messaging:inbox_conversation', kwargs={'pk': conversation.pk})) self.assertEqual(response.status_code, 200) self.assertContains(response, 'Hello') self.assertNotContains(response, 'Close chat') def test_unauthorized_user_cannot_access_conversation(self): outsider = User.objects.create_user( username='other', email='other@example.com', password='pass12345', role='seeker', ) conversation = get_or_create_conversation(self.application) self.client.login(username='other', password='pass12345') response = self.client.get(reverse('messaging:inbox_conversation', kwargs={'pk': conversation.pk})) self.assertEqual(response.status_code, 404) def test_ats_pipeline_has_message_link(self): self.client.login(username='employer1', password='pass12345') response = self.client.get(reverse('ats:pipeline')) self.assertEqual(response.status_code, 200) self.assertContains(response, f'/ats/applications/{self.application.pk}/#candidate-messages') class MessagingAPITestCase(TestCase): def setUp(self): self.employer = User.objects.create_user( username='api_employer', email='api_employer@example.com', password='pass12345', role='company', ) self.seeker_user = User.objects.create_user( username='api_seeker', email='api_seeker@example.com', password='pass12345', role='seeker', ) self.job = Job.objects.create( title='API Engineer', slug='api-engineer-test', company_name='API Co', company_slug='api-co', company_initials='AC', location='Lusaka, Lusaka, Zambia', job_type='remote', description='API work', responsibilities='Build APIs', requirements='REST', posted_by=self.employer, is_active=True, ) self.application = JobApplication.objects.create( job=self.job, applicant=self.seeker_user, ) self.conversation = get_or_create_conversation(self.application) self.employer_client = APIClient() self.seeker_client = APIClient() self.employer_client.credentials(HTTP_AUTHORIZATION=f'Token {Token.objects.create(user=self.employer).key}') self.seeker_client.credentials(HTTP_AUTHORIZATION=f'Token {Token.objects.create(user=self.seeker_user).key}') def test_api_start_send_close_reopen(self): start_resp = self.employer_client.post('/api/conversations/start/', {'application_id': self.application.pk}, format='json') self.assertEqual(start_resp.status_code, 201) send_resp = self.employer_client.post( f'/api/conversations/{self.conversation.pk}/send/', {'body': 'Interview invite'}, format='json', ) self.assertEqual(send_resp.status_code, 201) self.assertEqual(send_resp.data['body'], 'Interview invite') reply_resp = self.seeker_client.post( f'/api/conversations/{self.conversation.pk}/send/', {'body': 'Thanks, I am available'}, format='json', ) self.assertEqual(reply_resp.status_code, 201) close_resp = self.employer_client.post(f'/api/conversations/{self.conversation.pk}/close/') self.assertEqual(close_resp.status_code, 200) self.assertEqual(close_resp.data['status'], 'closed') blocked_resp = self.seeker_client.post( f'/api/conversations/{self.conversation.pk}/send/', {'body': 'One more thing'}, format='json', ) self.assertEqual(blocked_resp.status_code, 403) reopen_resp = self.employer_client.post(f'/api/conversations/{self.conversation.pk}/reopen/') self.assertEqual(reopen_resp.status_code, 200) self.assertEqual(reopen_resp.data['status'], 'open') def test_seeker_cannot_close_via_api(self): close_resp = self.seeker_client.post(f'/api/conversations/{self.conversation.pk}/close/') self.assertEqual(close_resp.status_code, 403) self.conversation.refresh_from_db() self.assertTrue(self.conversation.is_open) def test_conversation_list_includes_status(self): list_resp = self.employer_client.get('/api/conversations/') self.assertEqual(list_resp.status_code, 200) results = list_resp.data.get('results', list_resp.data) self.assertTrue(len(results) >= 1) self.assertEqual(results[0]['status'], 'open')