| import requests |
| import json |
| import logging |
| from datetime import datetime |
| import time |
| import os |
| import sys |
|
|
| |
| logging.basicConfig( |
| level=logging.INFO, |
| format='%(asctime)s - %(levelname)s - %(message)s', |
| handlers=[ |
| logging.FileHandler('endpoint_testing.log', encoding='utf-8'), |
| logging.StreamHandler() |
| ] |
| ) |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class EndpointTester: |
| def __init__(self, base_url='http://127.0.0.1:8000/api'): |
| self.base_url = base_url |
| self.results_dir = 'test_results1' |
| |
| os.makedirs(self.results_dir, exist_ok=True) |
|
|
| def test_translation_endpoint(self): |
| """Test the translation endpoint""" |
| endpoint = f'{self.base_url}/translate' |
| test_cases = [ |
| { |
| "user_message": "toilet me paani nahi aa rha hain", |
| "target_language": "en" |
| }, |
| { |
| "user_message": "AC not working properly", |
| "target_language": "hi" |
| }, |
| { |
| "user_message": "ਪੱਖਾ ਕੰਮ ਨਹੀਂ ਕਰ ਰਿਹਾ", |
| "target_language": "en" |
| } |
| ] |
|
|
| results = [] |
| for case in test_cases: |
| logger.info(f"\nTesting translation endpoint:") |
| logger.info(f"Input: {case}") |
| |
| try: |
| response = requests.post(endpoint, json=case) |
| result = response.json() |
| |
| logger.info(f"Status Code: {response.status_code}") |
| logger.info(f"Response: {result}") |
| |
| results.append({ |
| 'test_case': case, |
| 'status_code': response.status_code, |
| 'response': result |
| }) |
| |
| except Exception as e: |
| logger.error(f"Error testing translation: {str(e)}") |
| |
| time.sleep(1) |
| |
| return results |
|
|
| def test_sentiment_analysis_endpoint(self): |
| """Test the sentiment analysis endpoint""" |
| endpoint = f'{self.base_url}/analyze-sentiment' |
| test_cases = [ |
| { |
| "grievance_id": "G12347", |
| "text": "I am extremely frustrated with the constant power outages in my room." |
| }, |
| { |
| "grievance_id": "G12348", |
| "text": "Thank you for quickly fixing the water issue." |
| }, |
| { |
| "grievance_id": "G12349", |
| "text": "The AC has been making strange noises all night." |
| } |
| ] |
|
|
| results = [] |
| for case in test_cases: |
| logger.info(f"\nTesting sentiment analysis endpoint:") |
| logger.info(f"Input: {case}") |
| |
| try: |
| response = requests.post(endpoint, json=case) |
| result = response.json() |
| |
| logger.info(f"Status Code: {response.status_code}") |
| logger.info(f"Response: {result}") |
| |
| results.append({ |
| 'test_case': case, |
| 'status_code': response.status_code, |
| 'response': result |
| }) |
| |
| except Exception as e: |
| logger.error(f"Error testing sentiment analysis: {str(e)}") |
| |
| time.sleep(1) |
| |
| return results |
|
|
| def test_routing_endpoint(self): |
| """Test the grievance routing endpoint""" |
| endpoint = f'{self.base_url}/route-grievance' |
| test_cases = [ |
| { |
| "grievance_id": "G67890", |
| "category": "water_cooler", |
| "submission_timestamp": "2024-10-28T04:22:13Z", |
| "student_room_no": "148", |
| "hostel_name": "bh2", |
| "floor_number": 3, |
| "current_staff_status": [ |
| { |
| "staff_id": "S84148", |
| "department": "water_cooler", |
| "current_workload": 0, |
| "availability_status": "Unavailable", |
| "past_resolution_rate": 0.98 |
| }, |
| { |
| "staff_id": "S57369", |
| "department": "water_cooler", |
| "current_workload": 0, |
| "availability_status": "Available", |
| "past_resolution_rate": 0.96 |
| } |
| ], |
| "floor_metrics": { |
| "number_of_requests": 0, |
| "total_delays": 4 |
| }, |
| "availability_data": { |
| "staff_availability": [ |
| { |
| "staff_id": "S84148", |
| "time_slot": "08:00-12:00", |
| "availability_status": "Available" |
| } |
| ], |
| "student_availability": [ |
| { |
| "student_id": "STU200", |
| "time_slot": "16:00-20:00", |
| "availability_status": "Available" |
| } |
| ] |
| } |
| }, |
| { |
| "grievance_id": "G67891", |
| "category": "carpenter", |
| "submission_timestamp": "2024-10-28T04:18:13Z", |
| "student_room_no": "213", |
| "hostel_name": "bh1", |
| "floor_number": 2, |
| "current_staff_status": [ |
| { |
| "staff_id": "S52775", |
| "department": "carpenter", |
| "current_workload": 5, |
| "availability_status": "Unavailable", |
| "past_resolution_rate": 0.98 |
| }, |
| { |
| "staff_id": "S24943", |
| "department": "carpenter", |
| "current_workload": 3, |
| "availability_status": "Unavailable", |
| "past_resolution_rate": 0.85 |
| } |
| ], |
| "floor_metrics": { |
| "number_of_requests": 29, |
| "total_delays": 4 |
| }, |
| "availability_data": { |
| "staff_availability": [ |
| { |
| "staff_id": "S52775", |
| "time_slot": "12:00-16:00", |
| "availability_status": "Unavailable" |
| } |
| ], |
| "student_availability": [ |
| { |
| "student_id": "STU201", |
| "time_slot": "08:00-12:00", |
| "availability_status": "Available" |
| } |
| ] |
| } |
| }, |
| { |
| "grievance_id": "G67892", |
| "category": "electricity", |
| "submission_timestamp": "2024-10-28T04:00:13Z", |
| "student_room_no": "275", |
| "hostel_name": "bh3", |
| "floor_number": 2, |
| "current_staff_status": [ |
| { |
| "staff_id": "S33368", |
| "department": "electricity", |
| "current_workload": 1, |
| "availability_status": "Available", |
| "past_resolution_rate": 0.95 |
| }, |
| { |
| "staff_id": "S20522", |
| "department": "electricity", |
| "current_workload": 5, |
| "availability_status": "Available", |
| "past_resolution_rate": 0.87 |
| } |
| ], |
| "floor_metrics": { |
| "number_of_requests": 2, |
| "total_delays": 3 |
| }, |
| "availability_data": { |
| "staff_availability": [ |
| { |
| "staff_id": "S33368", |
| "time_slot": "12:00-16:00", |
| "availability_status": "Unavailable" |
| } |
| ], |
| "student_availability": [ |
| { |
| "student_id": "STU202", |
| "time_slot": "08:00-12:00", |
| "availability_status": "Unavailable" |
| } |
| ] |
| } |
| } |
| ] |
|
|
| results = [] |
| for case in test_cases: |
| logger.info(f"\nTesting routing endpoint:") |
| logger.info(f"Input: {case}") |
| |
| try: |
| response = requests.post(endpoint, json=case) |
| result = response.json() |
| |
| logger.info(f"Status Code: {response.status_code}") |
| logger.info(f"Response: {result}") |
| |
| results.append({ |
| 'test_case': case, |
| 'status_code': response.status_code, |
| 'response': result |
| }) |
| |
| except Exception as e: |
| logger.error(f"Error testing routing: {str(e)}") |
| |
| time.sleep(1) |
| |
| return results |
|
|
| def test_job_recommendation_endpoint(self): |
| """Test the job recommendation endpoint""" |
| endpoint = f'{self.base_url}/recommend-job' |
| test_cases = [ |
| { |
| "job_id": "J60000", |
| "type": "IT", |
| "description": "IT issue in room 358.", |
| "urgency_level": "Low", |
| "location": { |
| "hostel_name": "Hostel B", |
| "floor_number": 3, |
| "room_number": "358" |
| }, |
| "workers": [ |
| { |
| "worker_id": "W97053", |
| "department": "IT", |
| "current_workload": 3, |
| "availability_status": "Available", |
| "job_success_rate": 0.95, |
| "current_location": { |
| "hostel_name": "Hostel A", |
| "floor_number": 3, |
| "room_number": "468" |
| } |
| }, |
| { |
| "worker_id": "W97054", |
| "department": "IT", |
| "current_workload": 1, |
| "availability_status": "Available", |
| "job_success_rate": 0.92, |
| "current_location": { |
| "hostel_name": "Hostel B", |
| "floor_number": 3, |
| "room_number": "360" |
| } |
| } |
| ] |
| } |
| ] |
|
|
| results = [] |
| for case in test_cases: |
| logger.info(f"\nTesting job recommendation endpoint:") |
| logger.info(f"Input: {case}") |
| |
| try: |
| response = requests.post(endpoint, json=case) |
| result = response.json() |
| |
| logger.info(f"Status Code: {response.status_code}") |
| logger.info(f"Response: {result}") |
| |
| results.append({ |
| 'test_case': case, |
| 'status_code': response.status_code, |
| 'response': result |
| }) |
| |
| except Exception as e: |
| logger.error(f"Error testing job recommendation: {str(e)}") |
| |
| time.sleep(1) |
| |
| return results |
|
|
| def test_health_endpoint(self): |
| """Test the health check endpoint""" |
| endpoint = f'{self.base_url}/health' |
| |
| try: |
| response = requests.get(endpoint) |
| result = response.json() |
| |
| logger.info("\nTesting health endpoint:") |
| logger.info(f"Status Code: {response.status_code}") |
| logger.info(f"Response: {result}") |
| |
| return { |
| 'status_code': response.status_code, |
| 'response': result |
| } |
| |
| except Exception as e: |
| logger.error(f"Error testing health endpoint: {str(e)}") |
| return None |
|
|
| def run_all_tests(self): |
| """Run all endpoint tests""" |
| try: |
| |
| health_result = self.test_health_endpoint() |
| if not health_result or health_result['status_code'] != 200: |
| logger.error("Health check failed. Skipping other tests.") |
| return |
| |
| |
| results = { |
| 'timestamp': datetime.utcnow().isoformat(), |
| 'health_check': health_result, |
| 'translation_tests': self.test_translation_endpoint(), |
| 'sentiment_analysis_tests': self.test_sentiment_analysis_endpoint(), |
| 'routing_tests': self.test_routing_endpoint(), |
| 'job_recommendation_tests': self.test_job_recommendation_endpoint() |
| } |
| |
| |
| try: |
| results_file = os.path.join(self.results_dir, 'endpoint_test_results.json') |
| with open(results_file, 'w', encoding='utf-8') as f: |
| json.dump(results, f, indent=2, ensure_ascii=False) |
| logger.info(f"Test results saved to {results_file}") |
| except Exception as e: |
| logger.error(f"Error saving results: {str(e)}") |
| |
| |
| logger.info("All tests completed successfully") |
| return results |
| |
| except Exception as e: |
| logger.error(f"Error running tests: {str(e)}") |
| return { |
| 'timestamp': datetime.utcnow().isoformat(), |
| 'error': str(e), |
| 'completed_tests': [] |
| } |
|
|
| def main(): |
| try: |
| tester = EndpointTester() |
| results = tester.run_all_tests() |
| |
| if not results: |
| logger.error("No test results available") |
| return |
| |
| |
| print("\nTest Results Summary:") |
| for test_type, test_results in results.items(): |
| if test_type in ['timestamp', 'error']: |
| continue |
| |
| print(f"\n{test_type.replace('_', ' ').title()}:") |
| if isinstance(test_results, list): |
| success = sum(1 for r in test_results if r['status_code'] == 200) |
| total = len(test_results) |
| print(f"Total: {total}") |
| print(f"Successful: {success}") |
| print(f"Failed: {total - success}") |
| else: |
| status = test_results.get('status_code') == 200 if test_results else False |
| print(f"Status: {'Success' if status else 'Failed'}") |
| |
| except Exception as e: |
| logger.error(f"Error in main: {str(e)}") |
| raise |
|
|
| if __name__ == "__main__": |
| main() |
|
|
