kaauh_ats/recruitment/tests_advanced.py

1079 lines
38 KiB
Python

"""
Advanced test cases for the recruitment application.
These tests cover complex scenarios, API integrations, and edge cases.
"""
from django.test import TestCase, Client, TransactionTestCase
from django.contrib.auth.models import User, Group
from django.urls import reverse
from django.utils import timezone
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.exceptions import ValidationError
from django.db import IntegrityError
from django.db.models import ProtectedError, Q
from django.test.utils import override_settings
from django.core.management import call_command
from django.conf import settings
from unittest.mock import patch, MagicMock, Mock
from datetime import datetime, time, timedelta, date
import json
import os
import tempfile
from io import BytesIO
from PIL import Image
from .models import (
JobPosting, Candidate, ZoomMeeting, FormTemplate, FormStage, FormField,
FormSubmission, FieldResponse, InterviewSchedule, ScheduledInterview,
TrainingMaterial, Source, HiringAgency, Profile, MeetingComment, JobPostingImage,
BreakTime
)
from .forms import (
JobPostingForm, CandidateForm, ZoomMeetingForm, MeetingCommentForm,
CandidateStageForm, InterviewScheduleForm, BreakTimeFormSet
)
from .views import (
ZoomMeetingListView, ZoomMeetingCreateView, job_detail, candidate_screening_view,
candidate_exam_view, candidate_interview_view, submit_form, api_schedule_candidate_meeting,
schedule_interviews_view, confirm_schedule_interviews_view, _handle_preview_submission,
_handle_confirm_schedule, _handle_get_request
)
from .views_frontend import CandidateListView, JobListView, JobCreateView
from .utils import (
create_zoom_meeting, delete_zoom_meeting, update_zoom_meeting,
get_zoom_meeting_details, get_candidates_from_request,
get_available_time_slots
)
from .zoom_api import ZoomAPIError
class AdvancedModelTests(TestCase):
"""Advanced model tests with complex scenarios"""
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
is_staff=True
)
self.profile = Profile.objects.create(user=self.user)
self.job = JobPosting.objects.create(
title='Software Engineer',
department='IT',
job_type='FULL_TIME',
workplace_type='REMOTE',
location_country='Saudi Arabia',
description='Job description',
qualifications='Job qualifications',
created_by=self.user,
max_applications=10,
open_positions=2
)
def test_job_posting_complex_validation(self):
"""Test complex validation scenarios for JobPosting"""
# Test with valid data
valid_data = {
'title': 'Valid Job Title',
'department': 'IT',
'job_type': 'FULL_TIME',
'workplace_type': 'REMOTE',
'location_city': 'Riyadh',
'location_state': 'Riyadh',
'location_country': 'Saudi Arabia',
'description': 'Job description',
'qualifications': 'Job qualifications',
'salary_range': '5000-7000',
'application_deadline': '2025-12-31',
'max_applications': '100',
'open_positions': '2',
'hash_tags': '#hiring, #jobopening'
}
form = JobPostingForm(data=valid_data)
self.assertTrue(form.is_valid())
def test_job_posting_invalid_data_scenarios(self):
"""Test various invalid data scenarios for JobPosting"""
# Test empty title
invalid_data = {
'title': '',
'department': 'IT',
'job_type': 'FULL_TIME',
'workplace_type': 'REMOTE'
}
form = JobPostingForm(data=invalid_data)
self.assertFalse(form.is_valid())
self.assertIn('title', form.errors)
# Test invalid max_applications
invalid_data['title'] = 'Test Job'
invalid_data['max_applications'] = '0'
form = JobPostingForm(data=invalid_data)
self.assertFalse(form.is_valid())
# Test invalid hash_tags
invalid_data['max_applications'] = '100'
invalid_data['hash_tags'] = 'invalid hash tags without #'
form = JobPostingForm(data=invalid_data)
self.assertFalse(form.is_valid())
def test_candidate_stage_transition_validation(self):
"""Test advanced candidate stage transition validation"""
candidate = Candidate.objects.create(
first_name='John',
last_name='Doe',
email='john@example.com',
phone='1234567890',
job=self.job,
stage='Applied'
)
# Test valid transitions
valid_transitions = ['Exam', 'Interview', 'Offer']
for stage in valid_transitions:
candidate.stage = stage
candidate.save()
form = CandidateStageForm(data={'stage': stage}, candidate=candidate)
self.assertTrue(form.is_valid())
# Test invalid transition (e.g., from Offer back to Applied)
candidate.stage = 'Offer'
candidate.save()
form = CandidateStageForm(data={'stage': 'Applied'}, candidate=candidate)
# This should fail based on your STAGE_SEQUENCE logic
# Note: You'll need to implement can_transition_to method in Candidate model
def test_zoom_meeting_conflict_detection(self):
"""Test conflict detection for overlapping meetings"""
# Create a meeting
meeting1 = ZoomMeeting.objects.create(
topic='Meeting 1',
start_time=timezone.now() + timedelta(hours=1),
duration=60,
timezone='UTC',
join_url='https://zoom.us/j/123456789',
meeting_id='123456789'
)
# Try to create overlapping meeting (this logic would be in your view/service)
# This is a placeholder for actual conflict detection implementation
with self.assertRaises(ValidationError):
# This would trigger your conflict validation
pass
def test_form_template_integrity(self):
"""Test form template data integrity"""
template = FormTemplate.objects.create(
job=self.job,
name='Test Template',
created_by=self.user
)
# Create stages
stage1 = FormStage.objects.create(template=template, name='Stage 1', order=0)
stage2 = FormStage.objects.create(template=template, name='Stage 2', order=1)
# Create fields
field1 = FormField.objects.create(
stage=stage1, label='Field 1', field_type='text', order=0
)
field2 = FormField.objects.create(
stage=stage1, label='Field 2', field_type='email', order=1
)
# Test stage ordering
stages = template.stages.all()
self.assertEqual(stages[0], stage1)
self.assertEqual(stages[1], stage2)
# Test field ordering within stage
fields = stage1.fields.all()
self.assertEqual(fields[0], field1)
self.assertEqual(fields[1], field2)
def test_interview_schedule_complex_validation(self):
"""Test interview schedule validation with complex constraints"""
# Create candidates
candidate1 = Candidate.objects.create(
first_name='John', last_name='Doe', email='john@example.com',
phone='1234567890', job=self.job, stage='Interview'
)
candidate2 = Candidate.objects.create(
first_name='Jane', last_name='Smith', email='jane@example.com',
phone='9876543210', job=self.job, stage='Interview'
)
# Create schedule with valid data
schedule_data = {
'candidates': [candidate1.id, candidate2.id],
'start_date': date.today() + timedelta(days=1),
'end_date': date.today() + timedelta(days=7),
'working_days': [0, 1, 2, 3, 4], # Mon-Fri
'start_time': '09:00',
'end_time': '17:00',
'interview_duration': 60,
'buffer_time': 15,
'break_start_time': '12:00',
'break_end_time': '13:00'
}
form = InterviewScheduleForm(slug=self.job.slug, data=schedule_data)
self.assertTrue(form.is_valid())
def test_field_response_data_types(self):
"""Test different data types for field responses"""
# Create template and field
template = FormTemplate.objects.create(
job=self.job, name='Test Template', created_by=self.user
)
stage = FormStage.objects.create(template=template, name='Stage 1', order=0)
field = FormField.objects.create(
stage=stage, label='Test Field', field_type='text', order=0
)
# Create submission
submission = FormSubmission.objects.create(template=template)
# Test different value types
response = FieldResponse.objects.create(
submission=submission,
field=field,
value="Test string value"
)
self.assertEqual(response.display_value, "Test string value")
# Test list value (for checkbox/radio)
field.field_type = 'checkbox'
field.save()
response_checkbox = FieldResponse.objects.create(
submission=submission,
field=field,
value=["option1", "option2"]
)
self.assertEqual(response_checkbox.display_value, "option1, option2")
# Test file upload
file_content = b"Test file content"
uploaded_file = SimpleUploadedFile(
'test_file.pdf', file_content, content_type='application/pdf'
)
response_file = FieldResponse.objects.create(
submission=submission,
field=field,
uploaded_file=uploaded_file
)
self.assertTrue(response_file.is_file)
self.assertEqual(response_file.get_file_size, len(file_content))
class AdvancedViewTests(TestCase):
"""Advanced view tests with complex scenarios"""
def setUp(self):
self.client = Client()
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
is_staff=True
)
self.profile = Profile.objects.create(user=self.user)
self.job = JobPosting.objects.create(
title='Software Engineer',
department='IT',
job_type='FULL_TIME',
workplace_type='REMOTE',
location_country='Saudi Arabia',
description='Job description',
qualifications='Job qualifications',
created_by=self.user,
status='ACTIVE'
)
self.candidate = Candidate.objects.create(
first_name='John',
last_name='Doe',
email='john@example.com',
phone='1234567890',
job=self.job,
stage='Applied'
)
self.zoom_meeting = ZoomMeeting.objects.create(
topic='Interview with John Doe',
start_time=timezone.now() + timedelta(hours=1),
duration=60,
timezone='UTC',
join_url='https://zoom.us/j/123456789',
meeting_id='123456789'
)
def test_job_detail_with_multiple_candidates(self):
"""Test job detail view with multiple candidates at different stages"""
# Create more candidates at different stages
Candidate.objects.create(
first_name='Jane', last_name='Smith', email='jane@example.com',
phone='9876543210', job=self.job, stage='Exam'
)
Candidate.objects.create(
first_name='Bob', last_name='Johnson', email='bob@example.com',
phone='5555555555', job=self.job, stage='Interview'
)
Candidate.objects.create(
first_name='Alice', last_name='Brown', email='alice@example.com',
phone='4444444444', job=self.job, stage='Offer'
)
response = self.client.get(reverse('job_detail', kwargs={'slug': self.job.slug}))
self.assertEqual(response.status_code, 200)
# Check that counts are correct
self.assertContains(response, 'Total Applicants: 4')
self.assertContains(response, 'Applied: 1')
self.assertContains(response, 'Exam: 1')
self.assertContains(response, 'Interview: 1')
self.assertContains(response, 'Offer: 1')
def test_meeting_list_with_complex_filters(self):
"""Test meeting list view with multiple filter combinations"""
# Create meetings with different statuses and candidates
meeting2 = ZoomMeeting.objects.create(
topic='Interview with Jane Smith',
start_time=timezone.now() + timedelta(hours=2),
duration=60,
timezone='UTC',
join_url='https://zoom.us/j/987654321',
meeting_id='987654321',
status='started'
)
# Create scheduled interviews
ScheduledInterview.objects.create(
candidate=self.candidate,
job=self.job,
zoom_meeting=self.zoom_meeting,
interview_date=timezone.now().date(),
interview_time=time(10, 0),
status='scheduled'
)
ScheduledInterview.objects.create(
candidate=Candidate.objects.create(
first_name='Jane', last_name='Smith', email='jane@example.com',
phone='9876543210', job=self.job, stage='Interview'
),
job=self.job,
zoom_meeting=meeting2,
interview_date=timezone.now().date(),
interview_time=time(11, 0),
status='scheduled'
)
# Test combined filters
response = self.client.get(reverse('list_meetings'), {
'q': 'Interview',
'status': 'waiting',
'candidate_name': 'John'
})
self.assertEqual(response.status_code, 200)
def test_candidate_list_advanced_search(self):
"""Test candidate list view with advanced search functionality"""
# Create more candidates for testing
Candidate.objects.create(
first_name='Jane', last_name='Smith', email='jane@example.com',
phone='9876543210', job=self.job, stage='Exam'
)
Candidate.objects.create(
first_name='Bob', last_name='Johnson', email='bob@example.com',
phone='5555555555', job=self.job, stage='Interview'
)
# Test search by name
response = self.client.get(reverse('candidate_list'), {
'search': 'Jane'
})
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Jane Smith')
# Test search by email
response = self.client.get(reverse('candidate_list'), {
'search': 'bob@example.com'
})
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Bob Johnson')
# Test filter by job
response = self.client.get(reverse('candidate_list'), {
'job': self.job.slug
})
self.assertEqual(response.status_code, 200)
# Test filter by stage
response = self.client.get(reverse('candidate_list'), {
'stage': 'Exam'
})
self.assertEqual(response.status_code, 200)
def test_interview_scheduling_workflow(self):
"""Test the complete interview scheduling workflow"""
# Create candidates for scheduling
candidates = []
for i in range(3):
candidate = Candidate.objects.create(
first_name=f'Candidate{i}',
last_name=f'Test{i}',
email=f'candidate{i}@example.com',
phone=f'123456789{i}',
job=self.job,
stage='Interview'
)
candidates.append(candidate)
# Test GET request (initial form)
request = self.client.get(reverse('schedule_interviews', kwargs={'slug': self.job.slug}))
self.assertEqual(request.status_code, 200)
# Test POST request with preview
with patch('recruitment.views.get_available_time_slots') as mock_slots:
# Mock available time slots
mock_slots.return_value = [
{'date': date.today() + timedelta(days=1), 'time': '10:00'},
{'date': date.today() + timedelta(days=1), 'time': '11:00'},
{'date': date.today() + timedelta(days=1), 'time': '14:00'}
]
# Test _handle_preview_submission
self.client.login(username='testuser', password='testpass123')
post_data = {
'candidates': [c.pk for c in candidates],
'start_date': (date.today() + timedelta(days=1)).isoformat(),
'end_date': (date.today() + timedelta(days=7)).isoformat(),
'working_days': [0, 1, 2, 3, 4],
'start_time': '09:00',
'end_time': '17:00',
'interview_duration': '60',
'buffer_time': '15'
}
# This would normally be handled by the view, but we test the logic directly
# In a real test, you'd make a POST request to the view
request = self.client.post(
reverse('schedule_interviews', kwargs={'slug': self.job.slug}),
data=post_data
)
self.assertEqual(request.status_code, 200) # Should show preview
@patch('recruitment.views.create_zoom_meeting')
def test_meeting_creation_with_api_errors(self, mock_create):
"""Test meeting creation when API returns errors"""
# Test API error
mock_create.return_value = {
'status': 'error',
'message': 'Failed to create meeting'
}
self.client.login(username='testuser', password='testpass123')
data = {
'topic': 'Test Meeting',
'start_time': (timezone.now() + timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M'),
'duration': 60
}
response = self.client.post(reverse('create_meeting'), data)
# Should show error message
self.assertEqual(response.status_code, 200) # Form with error
def test_htmx_responses(self):
"""Test HTMX responses for partial updates"""
# Test HTMX request for candidate screening
response = self.client.get(
reverse('candidate_screening_view', kwargs={'slug': self.job.slug}),
HTTP_HX_REQUEST='true'
)
self.assertEqual(response.status_code, 200)
# Test HTMX request for meeting details
response = self.client.get(
reverse('meeting_details', kwargs={'slug': self.zoom_meeting.slug}),
HTTP_HX_REQUEST='true'
)
self.assertEqual(response.status_code, 200)
def test_bulk_operations(self):
"""Test bulk operations on candidates"""
# Create multiple candidates
candidates = []
for i in range(5):
candidate = Candidate.objects.create(
first_name=f'Bulk{i}',
last_name=f'Test{i}',
email=f'bulk{i}@example.com',
phone=f'123456789{i}',
job=self.job,
stage='Applied'
)
candidates.append(candidate)
# Test bulk status update
candidate_ids = [c.pk for c in candidates]
self.client.login(username='testuser', password='testpass123')
# This would be tested via a form submission
# For now, we test the view logic directly
request = self.client.post(
reverse('candidate_update_status', kwargs={'slug': self.job.slug}),
data={'candidate_ids': candidate_ids, 'mark_as': 'Exam'}
)
# Should redirect back to the view
self.assertEqual(request.status_code, 302)
# Verify candidates were updated
updated_count = Candidate.objects.filter(
pk__in=candidate_ids,
stage='Exam'
).count()
self.assertEqual(updated_count, len(candidates))
class AdvancedFormTests(TestCase):
"""Advanced form tests with complex validation scenarios"""
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
is_staff=True
)
self.job = JobPosting.objects.create(
title='Software Engineer',
department='IT',
job_type='FULL_TIME',
workplace_type='REMOTE',
location_country='Saudi Arabia',
description='Job description',
qualifications='Job qualifications',
created_by=self.user
)
def test_complex_form_validation_scenarios(self):
"""Test complex validation scenarios for forms"""
# Test JobPostingForm with all field types
complex_data = {
'title': 'Senior Software Engineer',
'department': 'Engineering',
'job_type': 'FULL_TIME',
'workplace_type': 'HYBRID',
'location_city': 'Riyadh',
'location_state': 'Riyadh',
'location_country': 'Saudi Arabia',
'description': 'Detailed job description',
'qualifications': 'Detailed qualifications',
'salary_range': '8000-12000 SAR',
'benefits': 'Health insurance, annual leave',
'application_start_date': '2025-01-01',
'application_deadline': '2025-12-31',
'application_instructions': 'Submit your resume online',
'position_number': 'ENG-2025-001',
'reporting_to': 'Engineering Manager',
'joining_date': '2025-06-01',
'created_by': self.user.get_full_name(),
'open_positions': '3',
'hash_tags': '#tech, #engineering, #senior',
'max_applications': '200'
}
form = JobPostingForm(data=complex_data)
self.assertTrue(form.is_valid(), form.errors)
def test_form_dependency_validation(self):
"""Test validation for dependent form fields"""
# Test InterviewScheduleForm with dependent fields
schedule_data = {
'candidates': [], # Empty for now
'start_date': '2025-01-15',
'end_date': '2025-01-10', # Invalid: end_date before start_date
'working_days': [0, 1, 2, 3, 4],
'start_time': '09:00',
'end_time': '17:00',
'interview_duration': '60',
'buffer_time': '15'
}
form = InterviewScheduleForm(slug=self.job.slug, data=schedule_data)
self.assertFalse(form.is_valid())
self.assertIn('end_date', form.errors)
def test_file_upload_validation(self):
"""Test file upload validation in forms"""
# Test valid file upload
valid_file = SimpleUploadedFile(
'valid_resume.pdf',
b'%PDF-1.4\n% ...',
content_type='application/pdf'
)
candidate_data = {
'job': self.job.id,
'first_name': 'John',
'last_name': 'Doe',
'phone': '1234567890',
'email': 'john@example.com',
'resume': valid_file
}
form = CandidateForm(data=candidate_data, files=candidate_data)
self.assertTrue(form.is_valid())
# Test invalid file type (would need custom validator)
# This test depends on your actual file validation logic
def test_dynamic_form_fields(self):
"""Test forms with dynamically populated fields"""
# Test InterviewScheduleForm with dynamic candidate queryset
# Create candidates in Interview stage
candidates = []
for i in range(3):
candidate = Candidate.objects.create(
first_name=f'Interview{i}',
last_name=f'Candidate{i}',
email=f'interview{i}@example.com',
phone=f'123456789{i}',
job=self.job,
stage='Interview'
)
candidates.append(candidate)
# Form should only show Interview stage candidates
form = InterviewScheduleForm(slug=self.job.slug)
self.assertEqual(form.fields['candidates'].queryset.count(), 3)
for candidate in candidates:
self.assertIn(candidate, form.fields['candidates'].queryset)
class AdvancedIntegrationTests(TransactionTestCase):
"""Advanced integration tests covering multiple components"""
def setUp(self):
self.client = Client()
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
is_staff=True
)
self.profile = Profile.objects.create(user=self.user)
def test_complete_hiring_workflow(self):
"""Test the complete hiring workflow from job posting to hire"""
# 1. Create job
job = JobPosting.objects.create(
title='Product Manager',
department='Product',
job_type='FULL_TIME',
workplace_type='ON_SITE',
location_country='Saudi Arabia',
description='Product Manager job description',
qualifications='Product management experience',
created_by=self.user,
status='ACTIVE'
)
# 2. Create form template for applications
template = FormTemplate.objects.create(
job=job,
name='Product Manager Application',
created_by=self.user,
is_active=True
)
# 3. Create form stages and fields
personal_stage = FormStage.objects.create(
template=template,
name='Personal Information',
order=0
)
FormField.objects.create(
stage=personal_stage,
label='First Name',
field_type='text',
order=0,
required=True
)
FormField.objects.create(
stage=personal_stage,
label='Last Name',
field_type='text',
order=1,
required=True
)
FormField.objects.create(
stage=personal_stage,
label='Email',
field_type='email',
order=2,
required=True
)
experience_stage = FormStage.objects.create(
template=template,
name='Work Experience',
order=1
)
FormField.objects.create(
stage=experience_stage,
label='Years of Experience',
field_type='number',
order=0
)
# 4. Submit application
submission_data = {
'field_1': 'Sarah',
'field_2': 'Johnson',
'field_3': 'sarah@example.com',
'field_4': '5'
}
response = self.client.post(
reverse('application_submit', kwargs={'template_id': template.id}),
submission_data
)
self.assertEqual(response.status_code, 302) # Redirect to success page
# 5. Verify candidate was created
candidate = Candidate.objects.get(email='sarah@example.com')
self.assertEqual(candidate.stage, 'Applied')
self.assertEqual(candidate.job, job)
# 6. Move candidate to Exam stage
candidate.stage = 'Exam'
candidate.save()
# 7. Move candidate to Interview stage
candidate.stage = 'Interview'
candidate.save()
# 8. Create interview schedule
scheduled_interview = ScheduledInterview.objects.create(
candidate=candidate,
job=job,
interview_date=timezone.now().date() + timedelta(days=7),
interview_time=time(14, 0),
status='scheduled'
)
# 9. Create Zoom meeting
zoom_meeting = ZoomMeeting.objects.create(
topic=f'Interview: {job.title} with {candidate.name}',
start_time=timezone.now() + timedelta(days=7, hours=14),
duration=60,
timezone='UTC',
join_url='https://zoom.us/j/interview123',
meeting_id='interview123'
)
# 10. Assign meeting to interview
scheduled_interview.zoom_meeting = zoom_meeting
scheduled_interview.save()
# 11. Verify all relationships
self.assertEqual(candidate.scheduled_interviews.count(), 1)
self.assertEqual(zoom_meeting.interview, scheduled_interview)
self.assertEqual(job.candidates.count(), 1)
# 12. Complete hire process
candidate.stage = 'Offer'
candidate.save()
# 13. Verify final state
self.assertEqual(Candidate.objects.filter(stage='Offer').count(), 1)
def test_data_integrity_across_operations(self):
"""Test data integrity across multiple operations"""
# Create complex data structure
job = JobPosting.objects.create(
title='Data Scientist',
department='Analytics',
job_type='FULL_TIME',
workplace_type='REMOTE',
location_country='Saudi Arabia',
description='Data Scientist position',
created_by=self.user,
max_applications=5
)
# Create multiple candidates
candidates = []
for i in range(3):
candidate = Candidate.objects.create(
first_name=f'Data{i}',
last_name=f'Scientist{i}',
email=f'data{i}@example.com',
phone=f'123456789{i}',
job=job,
stage='Applied'
)
candidates.append(candidate)
# Create form template
template = FormTemplate.objects.create(
job=job,
name='Data Scientist Application',
created_by=self.user,
is_active=True
)
# Create submissions for candidates
for i, candidate in enumerate(candidates):
submission = FormSubmission.objects.create(
template=template,
applicant_name=f'{candidate.first_name} {candidate.last_name}',
applicant_email=candidate.email
)
# Create field responses
FieldResponse.objects.create(
submission=submission,
field=FormField.objects.create(
stage=FormStage.objects.create(template=template, name='Stage 1', order=0),
label='Test Field',
field_type='text'
),
value=f'Test response {i}'
)
# Verify data consistency
self.assertEqual(FormSubmission.objects.filter(template=template).count(), 3)
self.assertEqual(FieldResponse.objects.count(), 3)
# Test application limit
for i in range(3): # Try to add more candidates than limit
Candidate.objects.create(
first_name=f'Extra{i}',
last_name=f'Candidate{i}',
email=f'extra{i}@example.com',
phone=f'11111111{i}',
job=job,
stage='Applied'
)
# Verify that the job shows application limit warning
job.refresh_from_db()
self.assertTrue(job.is_application_limit_reached)
@patch('recruitment.views.create_zoom_meeting')
def test_zoom_integration_workflow(self, mock_create):
"""Test complete Zoom integration workflow"""
# Setup job and candidate
job = JobPosting.objects.create(
title='Remote Developer',
department='Engineering',
job_type='REMOTE',
created_by=self.user
)
candidate = Candidate.objects.create(
first_name='Remote',
last_name='Developer',
email='remote@example.com',
job=job,
stage='Interview'
)
# Mock successful Zoom meeting creation
mock_create.return_value = {
'status': 'success',
'meeting_details': {
'meeting_id': 'zoom123',
'join_url': 'https://zoom.us/j/zoom123',
'password': 'meeting123'
},
'zoom_gateway_response': {
'status': 'waiting',
'id': 'meeting_zoom123'
}
}
# Schedule meeting via API
with patch('recruitment.views.ScheduledInterview.objects.create') as mock_create_interview:
mock_create_interview.return_value = ScheduledInterview(
candidate=candidate,
job=job,
zoom_meeting=None,
interview_date=timezone.now().date(),
interview_time=time(15, 0),
status='scheduled'
)
response = self.client.post(
reverse('api_schedule_candidate_meeting',
kwargs={'job_slug': job.slug, 'candidate_pk': candidate.pk}),
data={
'start_time': (timezone.now() + timedelta(hours=1)).isoformat(),
'duration': 60
}
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'success')
# Verify Zoom API was called
mock_create.assert_called_once()
# Verify interview was created
mock_create_interview.assert_called_once()
def test_concurrent_operations(self):
"""Test handling of concurrent operations"""
# Create job
job = JobPosting.objects.create(
title='Concurrency Test',
department='Test',
created_by=self.user
)
# Create candidates
candidates = []
for i in range(10):
candidate = Candidate.objects.create(
first_name=f'Concurrent{i}',
last_name=f'Test{i}',
email=f'concurrent{i}@example.com',
job=job,
stage='Applied'
)
candidates.append(candidate)
# Test concurrent candidate updates
from concurrent.futures import ThreadPoolExecutor
def update_candidate(candidate_id, stage):
from django.test import TestCase
from django.db import transaction
from recruitment.models import Candidate
with transaction.atomic():
candidate = Candidate.objects.select_for_update().get(pk=candidate_id)
candidate.stage = stage
candidate.save()
# Update candidates concurrently
with ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(update_candidate, c.pk, 'Exam')
for c in candidates
]
for future in futures:
future.result()
# Verify all updates completed
self.assertEqual(Candidate.objects.filter(stage='Exam').count(), len(candidates))
class SecurityTests(TestCase):
"""Security-focused tests"""
def setUp(self):
self.client = Client()
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
is_staff=False
)
self.staff_user = User.objects.create_user(
username='staffuser',
email='staff@example.com',
password='testpass123',
is_staff=True
)
self.job = JobPosting.objects.create(
title='Security Test Job',
department='Security',
job_type='FULL_TIME',
created_by=self.staff_user
)
def test_unauthorized_access_control(self):
"""Test that unauthorized users cannot access protected resources"""
# Test regular user accessing staff-only functionality
self.client.login(username='testuser', password='testpass123')
# Access job list (should be accessible)
response = self.client.get(reverse('job_list'))
self.assertEqual(response.status_code, 200)
# Try to edit job (should be restricted based on your actual implementation)
response = self.client.get(reverse('job_update', kwargs={'slug': self.job.slug}))
# This depends on your actual access control implementation
# For now, we'll assume it redirects or shows 403
def test_csrf_protection(self):
"""Test CSRF protection on forms"""
# Test POST request without CSRF token (should fail)
self.client.login(username='staffuser', password='testpass123')
response = self.client.post(
reverse('create_meeting'),
data={
'topic': 'Test Meeting',
'start_time': (timezone.now() + timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M'),
'duration': 60
},
HTTP_X_CSRFTOKEN='invalid' # Invalid or missing CSRF token
)
# Should be blocked by Django's CSRF protection
# The exact behavior depends on your middleware setup
def test_sql_injection_prevention(self):
"""Test that forms prevent SQL injection"""
# Test SQL injection in form fields
malicious_input = "Robert'); DROP TABLE candidates;--"
form_data = {
'title': f'SQL Injection Test {malicious_input}',
'department': 'IT',
'job_type': 'FULL_TIME',
'workplace_type': 'REMOTE'
}
form = JobPostingForm(data=form_data)
# Form should still be valid (malicious input stored as text, not executed)
self.assertTrue(form.is_valid())
# The actual protection comes from Django's ORM parameterized queries
def test_xss_prevention(self):
"""Test that forms prevent XSS attacks"""
# Test XSS attempt in form fields
xss_script = '<script>alert("XSS")</script>'
form_data = {
'title': f'XSS Test {xss_script}',
'department': 'IT',
'job_type': 'FULL_TIME',
'workplace_type': 'REMOTE'
}
form = JobPostingForm(data=form_data)
self.assertTrue(form.is_valid())
# The actual protection should be in template rendering
# Test template rendering with potentially malicious content
job = JobPosting.objects.create(
title=f'XSS Test {xss_script}',
department='IT',
created_by=self.staff_user
)