Spaces:
Runtime error
Runtime error
| import requests | |
| import json | |
| import time | |
| import asyncio | |
| from typing import Dict, List | |
| import logging | |
| from concurrent.futures import ThreadPoolExecutor | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class RAGSystemTester: | |
| def __init__(self, base_url: str = "http://localhost:8000", bearer_token: str = None): | |
| self.base_url = base_url | |
| self.bearer_token = bearer_token or "dbbdb701cfc45d4041e22a03edbfc65753fe9d7b4b9ba1df4884e864f3bb934d" | |
| self.headers = { | |
| "Authorization": f"Bearer {self.bearer_token}", | |
| "Content-Type": "application/json" | |
| } | |
| self.executor = ThreadPoolExecutor(max_workers=3) | |
| def test_health_check(self) -> bool: | |
| """Test health check endpoint""" | |
| try: | |
| response = requests.get(f"{self.base_url}/health", timeout=10) | |
| if response.status_code == 200: | |
| data = response.json() | |
| print(f"✅ Health check passed: {data}") | |
| return True | |
| else: | |
| print(f"❌ Health check failed: Status {response.status_code}") | |
| logger.error(f"Health check failed with status: {response.status_code}") | |
| return False | |
| except Exception as e: | |
| print(f"❌ Health check error: {str(e)}") | |
| logger.error(f"Health check error: {str(e)}") | |
| return False | |
| def test_metrics_endpoint(self) -> bool: | |
| """Test metrics endpoint""" | |
| try: | |
| response = requests.get(f"{self.base_url}/metrics", headers=self.headers, timeout=10) | |
| if response.status_code == 200: | |
| data = response.json() | |
| print(f"✅ Metrics endpoint passed: {json.dumps(data, indent=2)}") | |
| return True | |
| else: | |
| print(f"❌ Metrics endpoint failed: Status {response.status_code}") | |
| logger.error(f"Metrics endpoint failed with status: {response.status_code}") | |
| return False | |
| except Exception as e: | |
| print(f"❌ Metrics endpoint error: {str(e)}") | |
| logger.error(f"Metrics endpoint error: {str(e)}") | |
| return False | |
| def test_sample_query(self) -> bool: | |
| """Test with the provided sample data""" | |
| sample_data = { | |
| "documents": "https://hackrx.blob.core.windows.net/assets/Arogya%20Sanjeevani%20Policy%20-%20CIN%20-%20U10200WB1906GOI001713%201.pdf?sv=2023-01-03&st=2025-07-21T08%3A29%3A02Z&se=2025-09-22T08%3A29%3A00Z&sr=b&sp=r&sig=nzrz1K9Iurt%2BBXom%2FB%2BMPTFMFP3PRnIvEsipAX10Ig4%3D", | |
| "questions": [ | |
| "What is the grace period for premium payment under the National Parivar Mediclaim Plus Policy?", | |
| "What is the waiting period for pre-existing diseases (PED) to be covered?", | |
| "Does this policy cover maternity expenses, and what are the conditions?", | |
| "What is the waiting period for cataract surgery?", | |
| "Are the medical expenses for an organ donor covered under this policy?", | |
| "What is the No Claim Discount (NCD) offered in this policy?", | |
| "Is there a benefit for preventive health check-ups?", | |
| "How does the policy define a 'Hospital'?", | |
| "What is the extent of coverage for AYUSH treatments?", | |
| "Are there any sub-limits on room rent and ICU charges for Plan A?" | |
| ] | |
| } | |
| try: | |
| print("🔄 Testing sample query...") | |
| start_time = time.time() | |
| response = requests.post( | |
| f"{self.base_url}/hackrx/run", | |
| headers=self.headers, | |
| json=sample_data, | |
| timeout=120 | |
| ) | |
| end_time = time.time() | |
| latency = end_time - start_time | |
| if response.status_code == 200: | |
| data = response.json() | |
| answers = data.get("answers", []) | |
| print(f"✅ Sample query successful (Latency: {latency:.2f}s)") | |
| print(f"📊 Received {len(answers)} answers") | |
| # Print all answers for validation | |
| for i, (question, answer) in enumerate(zip(sample_data['questions'], answers)): | |
| print(f"Q{i+1}: {question}") | |
| print(f"A{i+1}: {answer[:200]}..." if len(answer) > 200 else f"A{i+1}: {answer}") | |
| print("-" * 50) | |
| # Validate that we received answers for all questions | |
| if len(answers) == len(sample_data['questions']): | |
| print("✅ All questions answered") | |
| return True | |
| else: | |
| print(f"❌ Incomplete response: Expected {len(sample_data['questions'])} answers, got {len(answers)}") | |
| logger.warning(f"Incomplete response: Expected {len(sample_data['questions'])} answers, got {len(answers)}") | |
| return False | |
| else: | |
| print(f"❌ Sample query failed: Status {response.status_code}") | |
| print(f"Response: {response.text}") | |
| logger.error(f"Sample query failed: Status {response.status_code}, Response: {response.text}") | |
| return False | |
| except Exception as e: | |
| print(f"❌ Sample query error: {str(e)}") | |
| logger.error(f"Sample query error: {str(e)}") | |
| return False | |
| async def test_concurrent_queries(self, num_requests: int = 3) -> bool: | |
| """Test system under concurrent load""" | |
| async def make_request(): | |
| try: | |
| response = requests.post( | |
| f"{self.base_url}/hackrx/run", | |
| headers=self.headers, | |
| json={ | |
| "documents": "https://hackrx.blob.core.windows.net/assets/Arogya%20Sanjeevani%20Policy%20-%20CIN%20-%20U10200WB1906GOI001713%201.pdf?sv=2023-01-03&st=2025-07-21T08%3A29%3A02Z&se=2025-09-22T08%3A29%3A00Z&sr=b&sp=r&sig=nzrz1K9Iurt%2BBXom%2FB%2BMPTFMFP3PRnIvEsipAX10Ig4%3D", | |
| "questions": ["What is the grace period for premium payment?"] | |
| }, | |
| timeout=60 | |
| ) | |
| return response.status_code == 200 | |
| except Exception as e: | |
| logger.error(f"Concurrent query error: {str(e)}") | |
| return False | |
| print(f"🔄 Testing {num_requests} concurrent queries...") | |
| tasks = [make_request() for _ in range(num_requests)] | |
| results = await asyncio.gather(*tasks, return_exceptions=True) | |
| success_count = sum(1 for result in results if result is True) | |
| print(f"✅ Concurrent test completed: {success_count}/{num_requests} successful") | |
| return success_count == num_requests | |
| def test_invalid_token(self) -> bool: | |
| """Test authentication with invalid token""" | |
| try: | |
| invalid_headers = { | |
| "Authorization": "Bearer invalid_token", | |
| "Content-Type": "application/json" | |
| } | |
| response = requests.post( | |
| f"{self.base_url}/hackrx/run", | |
| headers=invalid_headers, | |
| json={ | |
| "documents": "https://hackrx.blob.core.windows.net/assets/Arogya%20Sanjeevani%20Policy%20-%20CIN%20-%20U10200WB1906GOI001713%201.pdf?sv=2023-01-03&st=2025-07-21T08%3A29%3A02Z&se=2025-09-22T08%3A29%3A00Z&sr=b&sp=r&sig=nzrz1K9Iurt%2BBXom%2FB%2BMPTFMFP3PRnIvEsipAX10Ig4%3D", | |
| "questions": ["Test question"] | |
| }, | |
| timeout=10 | |
| ) | |
| if response.status_code == 401: | |
| print("✅ Invalid token test passed: Correctly rejected") | |
| return True | |
| else: | |
| print(f"❌ Invalid token test failed: Expected 401, got {response.status_code}") | |
| logger.warning(f"Invalid token test failed: Expected 401, got {response.status_code}") | |
| return False | |
| except Exception as e: | |
| print(f"❌ Invalid token test error: {str(e)}") | |
| logger.error(f"Invalid token test error: {str(e)}") | |
| return False | |
| def test_invalid_url(self) -> bool: | |
| """Test with invalid document URL""" | |
| try: | |
| response = requests.post( | |
| f"{self.base_url}/hackrx/run", | |
| headers=self.headers, | |
| json={ | |
| "documents": "https://invalid-url-that-does-not-exist.com/fake.pdf", # Actually invalid URL | |
| "questions": ["Test question"] | |
| }, | |
| timeout=30 | |
| ) | |
| # Accept either 400 or 500 as valid error responses for invalid URLs | |
| if response.status_code in [400, 500]: | |
| print("✅ Invalid URL test passed: Correctly handled") | |
| return True | |
| else: | |
| print(f"❌ Invalid URL test failed: Expected 400/500, got {response.status_code}") | |
| logger.warning(f"Invalid URL test failed: Expected 400/500, got {response.status_code}") | |
| return False | |
| except Exception as e: | |
| print(f"❌ Invalid URL test error: {str(e)}") | |
| logger.error(f"Invalid URL test error: {str(e)}") | |
| return False | |
| async def run_all_tests(self): | |
| """Run all test cases""" | |
| print("🚀 Starting RAG System Tests") | |
| print("=" * 50) | |
| results = { | |
| "health_check": self.test_health_check(), | |
| "metrics_endpoint": self.test_metrics_endpoint(), | |
| "sample_query": self.test_sample_query(), | |
| "concurrent_queries": await self.test_concurrent_queries(), | |
| "invalid_token": self.test_invalid_token(), | |
| "invalid_url": self.test_invalid_url() | |
| } | |
| print("\n📊 Test Summary") | |
| print("=" * 50) | |
| passed = sum(1 for result in results.values() if result) | |
| total = len(results) | |
| for test_name, passed in results.items(): | |
| status = "✅ PASSED" if passed else "❌ FAILED" | |
| print(f"{test_name}: {status}") | |
| print(f"\n🎯 Overall: {passed}/{total} tests passed") | |
| return passed == total | |
| def main(): | |
| tester = RAGSystemTester() | |
| asyncio.run(tester.run_all_tests()) | |
| if __name__ == "__main__": | |
| main() |