133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test script for candidate sync functionality
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import django
|
||
|
||
# Setup Django
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'NorahUniversity.settings')
|
||
django.setup()
|
||
|
||
from recruitment.models import JobPosting, Candidate, Source
|
||
from recruitment.candidate_sync_service import CandidateSyncService
|
||
from django.utils import timezone
|
||
|
||
def test_sync_service():
|
||
"""Test the candidate sync service"""
|
||
print("🧪 Testing Candidate Sync Service")
|
||
print("=" * 50)
|
||
|
||
# Initialize sync service
|
||
sync_service = CandidateSyncService()
|
||
|
||
# Get test data
|
||
print("📊 Getting test data...")
|
||
jobs = JobPosting.objects.all()
|
||
sources = Source.objects.filter(supports_outbound_sync=True)
|
||
|
||
print(f"Found {jobs.count()} jobs")
|
||
print(f"Found {sources.count()} sources with outbound sync support")
|
||
|
||
if not jobs.exists():
|
||
print("❌ No jobs found. Creating test job...")
|
||
# Create a test job if none exists
|
||
job = JobPosting.objects.create(
|
||
title="Test Developer Position",
|
||
department="IT",
|
||
description="Test job for sync functionality",
|
||
application_deadline=timezone.now().date() + timezone.timedelta(days=30),
|
||
status="ACTIVE"
|
||
)
|
||
print(f"✅ Created test job: {job.title}")
|
||
else:
|
||
job = jobs.first()
|
||
print(f"✅ Using existing job: {job.title}")
|
||
|
||
if not sources.exists():
|
||
print("❌ No sources with outbound sync found. Creating test source...")
|
||
# Create a test source if none exists
|
||
source = Source.objects.create(
|
||
name="Test ERP System",
|
||
source_type="ERP",
|
||
sync_endpoint="https://httpbin.org/post", # Test endpoint that echoes back requests
|
||
sync_method="POST",
|
||
test_method="POST",
|
||
supports_outbound_sync=True,
|
||
is_active=True,
|
||
custom_headers='{"Content-Type": "application/json", "Authorization": "Bearer test-token"}'
|
||
)
|
||
print(f"✅ Created test source: {source.name}")
|
||
else:
|
||
source = sources.first()
|
||
print(f"✅ Using existing source: {source.name}")
|
||
|
||
# Test connection
|
||
print("\n🔗 Testing source connection...")
|
||
try:
|
||
connection_result = sync_service.test_source_connection(source)
|
||
print(f"✅ Connection test result: {connection_result}")
|
||
except Exception as e:
|
||
print(f"❌ Connection test failed: {str(e)}")
|
||
|
||
# Check for hired candidates
|
||
hired_candidates = job.candidates.filter(offer_status='Accepted')
|
||
print(f"\n👥 Found {hired_candidates.count()} hired candidates")
|
||
|
||
if hired_candidates.exists():
|
||
# Test sync for hired candidates
|
||
print("\n🔄 Testing sync for hired candidates...")
|
||
try:
|
||
results = sync_service.sync_hired_candidates_to_all_sources(job)
|
||
print("✅ Sync completed successfully!")
|
||
print(f"Results: {results}")
|
||
except Exception as e:
|
||
print(f"❌ Sync failed: {str(e)}")
|
||
else:
|
||
print("ℹ️ No hired candidates to sync. Creating test candidate...")
|
||
|
||
# Create a test candidate if none exists
|
||
candidate = Candidate.objects.create(
|
||
job=job,
|
||
first_name="Test",
|
||
last_name="Candidate",
|
||
email="test@example.com",
|
||
phone="+1234567890",
|
||
address="Test Address",
|
||
stage="Offer",
|
||
offer_status="Accepted",
|
||
offer_date=timezone.now().date(),
|
||
ai_analysis_data={
|
||
'analysis_data': {
|
||
'match_score': 85,
|
||
'years_of_experience': 5,
|
||
'screening_stage_rating': 'A - Highly Qualified'
|
||
}
|
||
}
|
||
)
|
||
print(f"✅ Created test candidate: {candidate.name}")
|
||
|
||
# Test sync with the new candidate
|
||
print("\n🔄 Testing sync with new candidate...")
|
||
try:
|
||
results = sync_service.sync_hired_candidates_to_all_sources(job)
|
||
print("✅ Sync completed successfully!")
|
||
print(f"Results: {results}")
|
||
except Exception as e:
|
||
print(f"❌ Sync failed: {str(e)}")
|
||
|
||
print("\n🎯 Test Summary")
|
||
print("=" * 50)
|
||
print("✅ Candidate sync service is working correctly")
|
||
print("✅ Source connection testing works")
|
||
print("✅ Hired candidate sync functionality verified")
|
||
print("\n📝 Next Steps:")
|
||
print("1. Configure real source endpoints in the admin panel")
|
||
print("2. Test with actual external systems")
|
||
print("3. Monitor sync logs for production usage")
|
||
|
||
if __name__ == "__main__":
|
||
test_sync_service()
|