TimeLooker MCP Server

""" Local testing script for TimeLooker Lambda function. Simulates AWS Lambda invocations for development and testing. """ import json import time import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from scripts.lambda_function import lambda_handler class LambdaTester: """ Simulates AWS Lambda environment for local testing. """ def __init__(self): self.test_results = [] def invoke_lambda(self, event, description=""): """ Invoke lambda function and track results. Args: event: Event dictionary to pass to lambda description: Description of the test """ print(f"\n{'='*60}") print(f"TEST: {description}") print(f"Event: {json.dumps(event, indent=2)}") print(f"{'='*60}") start_time = time.time() try: response = lambda_handler(event, None) execution_time = time.time() - start_time print(f"✓ Lambda executed successfully in {execution_time:.2f}s") print(f"Response: {json.dumps(response, indent=2)}") # Determine if this is an expected failure based on test description expected_failure = any(phrase in description.lower() for phrase in [ 'should fail', 'should return error', 'non-existent', 'invalid', 'missing fields' ]) status_code = response.get('statusCode', 500) # For expected failures, success means getting an error status code (4xx or 5xx) # For normal tests, success means getting a success status code (2xx or 3xx) if expected_failure: success = status_code >= 400 else: success = status_code < 400 self.test_results.append({ 'description': description, 'event': event, 'response': response, 'execution_time': execution_time, 'success': success, 'expected_failure': expected_failure, 'status_code': status_code }) return response except Exception as e: execution_time = time.time() - start_time print(f"✗ Lambda execution failed in {execution_time:.2f}s") print(f"Error: {str(e)}") self.test_results.append({ 'description': description, 'event': event, 'error': str(e), 'execution_time': execution_time, 'success': False, 'expected_failure': False, 'status_code': 500 }) return None def run_full_test_suite(self): """ Run a comprehensive test suite that simulates real usage. """ print("Starting TimeLooker Lambda Test Suite") print("=====================================") # Test 1: List tasks (should be empty initially) self.invoke_lambda( {"action": "list_tasks"}, "List tasks (initial - should be empty)" ) # Test 2: Create a new task create_response = self.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "AI Ethics and Safety openings fit for a PhD in Computer Science", "frequency_minutes": 30, "runtime_minutes": 120, "recipient_email": "user@example.com" } }, "Create new monitoring task" ) # Extract task ID for subsequent tests task_id = None if create_response and create_response.get('statusCode') == 200: body = json.loads(create_response['body']) task_id = body.get('task_id') # Test 3: List tasks (should now show the created task) self.invoke_lambda( {"action": "list_tasks"}, "List tasks (after creating one)" ) if task_id: # Test 4: Execute search for the created task self.invoke_lambda( { "action": "execute_search", "task_id": task_id }, f"Execute search for task {task_id} (first run)" ) # Test 5: Execute search again (should detect duplicates) self.invoke_lambda( { "action": "execute_search", "task_id": task_id }, f"Execute search for task {task_id} (second run - should detect duplicates)" ) # Test 6: Try to execute search for non-existent task self.invoke_lambda( { "action": "execute_search", "task_id": 99999 }, "Execute search for non-existent task (should fail)" ) # Test 7: Invalid action self.invoke_lambda( {"action": "invalid_action"}, "Invalid action (should return error)" ) # Test 8: Create task with missing fields self.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "Incomplete task" # Missing required fields } }, "Create task with missing fields (should fail)" ) # Test 9: Create another task with different parameters self.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "New sneaker releases from Nike and Adidas", "frequency_minutes": 60, "runtime_minutes": 180, "recipient_email": "sneaker.fan@example.com" } }, "Create second task (different domain)" ) # Test 10: Final task list self.invoke_lambda( {"action": "list_tasks"}, "Final task list (should show multiple tasks)" ) self.print_test_summary() def run_frequency_test(self): """ Test the frequency checking mechanism. """ print("\nTesting frequency mechanism...") # Create a task with short frequency for testing create_response = self.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "Test frequency task", "frequency_minutes": 1, # Very short for testing "runtime_minutes": 60, "recipient_email": "test@example.com" } }, "Create task for frequency testing" ) if create_response and create_response.get('statusCode') == 200: body = json.loads(create_response['body']) task_id = body.get('task_id') # First execution should work self.invoke_lambda( {"action": "execute_search", "task_id": task_id}, "First execution (should work)" ) # Immediate second execution should be skipped self.invoke_lambda( {"action": "execute_search", "task_id": task_id}, "Immediate second execution (should be skipped due to frequency)" ) print("Waiting 65 seconds to test frequency mechanism...") time.sleep(65) # Third execution after waiting should work self.invoke_lambda( {"action": "execute_search", "task_id": task_id}, "Third execution after waiting (should work)" ) def print_test_summary(self): """ Print a summary of all test results. """ print(f"\n{'='*60}") print("TEST SUMMARY") print(f"{'='*60}") total_tests = len(self.test_results) successful_tests = sum(1 for result in self.test_results if result['success']) failed_tests = total_tests - successful_tests # Count expected vs unexpected failures unexpected_failures = sum(1 for result in self.test_results if not result['success'] and not result.get('expected_failure', False)) expected_failures = sum(1 for result in self.test_results if result['success'] and result.get('expected_failure', False)) print(f"Total tests: {total_tests}") print(f"Successful: {successful_tests} (including {expected_failures} expected failures)") print(f"Failed: {failed_tests}") print(f"Unexpected failures: {unexpected_failures}") print(f"Success rate: {(successful_tests/total_tests)*100:.1f}%") if unexpected_failures > 0: print(f"\nUnexpected failures:") for result in self.test_results: if not result['success'] and not result.get('expected_failure', False): print(f" ✗ {result['description']}") if 'error' in result: print(f" Error: {result['error']}") else: print(f" Status: {result.get('status_code', 'Unknown')}") if expected_failures > 0: print(f"\nExpected failures (working correctly):") for result in self.test_results: if result['success'] and result.get('expected_failure', False): print(f" ✓ {result['description']} (Status: {result.get('status_code', 'Unknown')})") # Show execution times avg_time = sum(result['execution_time'] for result in self.test_results) / total_tests max_time = max(result['execution_time'] for result in self.test_results) min_time = min(result['execution_time'] for result in self.test_results) print(f"\nExecution times:") print(f" Average: {avg_time:.2f}s") print(f" Max: {max_time:.2f}s") print(f" Min: {min_time:.2f}s") def main(): """ Main function to run tests. """ import argparse parser = argparse.ArgumentParser(description="Test TimeLooker Lambda function locally") parser.add_argument("--full", action="store_true", help="Run full test suite") parser.add_argument("--frequency", action="store_true", help="Run frequency test (takes ~2 minutes)") parser.add_argument("--quick", action="store_true", help="Run quick basic tests") args = parser.parse_args() tester = LambdaTester() if args.full: tester.run_full_test_suite() elif args.frequency: tester.run_frequency_test() elif args.quick: # Quick test - just create and list tester.invoke_lambda({"action": "list_tasks"}, "Quick test - list tasks") create_response = tester.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "Quick test task", "frequency_minutes": 60, "runtime_minutes": 120, "recipient_email": "test@example.com" } }, "Quick test - create task" ) tester.print_test_summary() else: # Default: run basic functionality test print("Running basic functionality test...") print("Use --full for comprehensive testing, --quick for minimal testing") tester.invoke_lambda({"action": "list_tasks"}, "List tasks") tester.invoke_lambda( { "action": "create_task", "task_config": { "task_description": "Test monitoring task", "frequency_minutes": 30, "runtime_minutes": 120, "recipient_email": "test@example.com" } }, "Create test task" ) tester.print_test_summary() if __name__ == "__main__": main()