|
import sys |
|
import os |
|
from pathlib import Path |
|
|
|
|
|
project_root = Path(__file__).parent |
|
sys.path.append(str(project_root)) |
|
|
|
from core.redis_client import redis_client |
|
|
|
def main(): |
|
"""Verify Redis connection with exact configuration""" |
|
print("Verifying Redis connection with exact configuration...") |
|
|
|
|
|
success = redis_client.test_basic_connection() |
|
|
|
if success: |
|
print("\nβ
Redis connection verified successfully!") |
|
print("The exact configuration from Redis Cloud works correctly.") |
|
else: |
|
print("\nβ Redis connection verification failed!") |
|
print("Please check your configuration and network connectivity.") |
|
return 1 |
|
|
|
|
|
print("\nTesting application Redis client...") |
|
client = redis_client.get_client() |
|
|
|
if client is None: |
|
print("β Application Redis client is None") |
|
return 1 |
|
|
|
try: |
|
client.ping() |
|
print("β
Application Redis client ping successful") |
|
|
|
|
|
client.set('app_test_key', 'app_test_value') |
|
value = client.get('app_test_key') |
|
client.delete('app_test_key') |
|
|
|
if value == 'app_test_value': |
|
print("β
Set/Get operations work correctly") |
|
print("\nπ All Redis connection tests passed!") |
|
return 0 |
|
else: |
|
print("β Set/Get operations failed") |
|
return 1 |
|
|
|
except Exception as e: |
|
print(f"β Application Redis client test failed: {e}") |
|
return 1 |
|
|
|
if __name__ == "__main__": |
|
exit_code = main() |
|
sys.exit(exit_code) |
|
|