-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathtest_supabase_integration.py
More file actions
116 lines (87 loc) Β· 3.47 KB
/
test_supabase_integration.py
File metadata and controls
116 lines (87 loc) Β· 3.47 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
#!/usr/bin/env python3
"""
Test script to verify Supabase integration is working
"""
import os
import sys
from dotenv import load_dotenv
# Add the app directory to the Python path
sys.path.append(os.path.join(os.path.dirname(__file__), 'app'))
# Load environment variables
load_dotenv()
def test_database_adapter():
"""Test the database adapter"""
print("π§ͺ Testing Database Adapter...")
try:
from app.database_adapter import get_database_adapter
db_adapter = get_database_adapter()
print(f"β
Database adapter initialized (using {db_adapter.database_type})")
# Test getting hotels
hotels = db_adapter.get_hotels()
print(f"β
Retrieved {len(hotels)} hotels from database")
for hotel in hotels:
print(f" - Hotel: {hotel['hotel_name']} (ID: {hotel['id']})")
# Test authentication
if hotels:
first_hotel = hotels[0]
hotel_id = db_adapter.authenticate_hotel(first_hotel['hotel_name'], first_hotel['password'])
if hotel_id:
print(f"β
Authentication successful for hotel: {first_hotel['hotel_name']}")
else:
print(f"β Authentication failed for hotel: {first_hotel['hotel_name']}")
return True
except Exception as e:
print(f"β Database adapter test failed: {e}")
return False
def test_storage_adapter():
"""Test the storage adapter"""
print("\nπ§ͺ Testing Storage Adapter...")
try:
from app.storage_adapter import get_storage_adapter
storage_adapter = get_storage_adapter()
print(f"β
Storage adapter initialized (using {'Supabase' if storage_adapter.use_supabase else 'Local'})")
return True
except Exception as e:
print(f"β Storage adapter test failed: {e}")
return False
def test_supabase_connection():
"""Test direct Supabase connection"""
print("\nπ§ͺ Testing Supabase Connection...")
try:
from app.supabase_config import get_supabase_client
supabase = get_supabase_client()
print("β
Supabase client initialized")
# Test a simple query
result = supabase.table("hotels").select("id, hotel_name").limit(1).execute()
if result.data:
print(f"β
Supabase query successful: {result.data[0]}")
else:
print("βΉοΈ Supabase query returned no data (this is normal if no hotels exist)")
return True
except Exception as e:
print(f"β Supabase connection test failed: {e}")
return False
def main():
"""Main test function"""
print("π Starting Supabase Integration Tests...\n")
tests_passed = 0
total_tests = 3
# Test database adapter
if test_database_adapter():
tests_passed += 1
# Test storage adapter
if test_storage_adapter():
tests_passed += 1
# Test Supabase connection
if test_supabase_connection():
tests_passed += 1
print(f"\nπ Test Results: {tests_passed}/{total_tests} tests passed")
if tests_passed == total_tests:
print("π All tests passed! Supabase integration is working correctly.")
return True
else:
print("β Some tests failed. Please check the configuration.")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)