|
|
import os |
|
|
from dotenv import load_dotenv |
|
|
from supabase import create_client, Client |
|
|
from supabase_auth.errors import AuthApiError |
|
|
|
|
|
|
|
|
load_dotenv() |
|
|
|
|
|
|
|
|
SUPABASE_URL = os.environ.get("SUPABASE_URL") |
|
|
SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY") |
|
|
|
|
|
if not SUPABASE_URL or not SUPABASE_ANON_KEY: |
|
|
print("Error: Supabase URL and Anon Key must be set as environment variables in .env file.") |
|
|
exit(1) |
|
|
|
|
|
supabase_anon: Client = create_client(SUPABASE_URL, SUPABASE_ANON_KEY) |
|
|
|
|
|
SUPABASE_SERVICE_ROLE_KEY = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") |
|
|
if not SUPABASE_SERVICE_ROLE_KEY: |
|
|
print("Error: SUPABASE_SERVICE_ROLE_KEY must be set as an environment variable in .env file for admin operations.") |
|
|
exit(1) |
|
|
supabase_admin: Client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY) |
|
|
|
|
|
async def test_db_connection(): |
|
|
"""Tests a simple database connection by querying a non-existent table.""" |
|
|
print("\n--- Testing Database Connection ---") |
|
|
try: |
|
|
|
|
|
|
|
|
res = supabase_anon.table('non_existent_table').select('*').limit(1).execute() |
|
|
print("Database connection successful (or query executed without connection error).") |
|
|
print(f"Query result (expected empty or error): {res.data}") |
|
|
except Exception as e: |
|
|
print(f"Database connection test failed: {e}") |
|
|
print("Please ensure SUPABASE_URL is correct and your network allows access to Supabase.") |
|
|
|
|
|
async def test_user_registration(email, password): |
|
|
"""Tests user registration with Supabase.""" |
|
|
print(f"\n--- Testing User Registration for {email} ---") |
|
|
try: |
|
|
res = supabase_anon.auth.sign_up({ |
|
|
"email": email, |
|
|
"password": password |
|
|
}) |
|
|
if res.user: |
|
|
print(f"User registration successful for {res.user.email}! Please check your email for verification.") |
|
|
print(f"User ID: {res.user.id}") |
|
|
else: |
|
|
print("User registration failed: No user returned.") |
|
|
print(f"Supabase response: {res}") |
|
|
except AuthApiError as e: |
|
|
print(f"User registration failed (AuthApiError): {e.message}") |
|
|
except Exception as e: |
|
|
print(f"An unexpected error occurred during registration: {e}") |
|
|
print("Please ensure Supabase URL and Anon Key are correct, and Email Signups are enabled in Supabase Auth settings.") |
|
|
|
|
|
async def list_supabase_users(): |
|
|
"""Lists all users in Supabase.""" |
|
|
print("\n--- Listing Supabase Users ---") |
|
|
try: |
|
|
|
|
|
users_list = supabase_admin.auth.admin.list_users() |
|
|
|
|
|
if users_list: |
|
|
print(f"Found {len(users_list)} users:") |
|
|
for user in users_list: |
|
|
print(f" User ID: {user.id}, Email: {user.email}, Created At: {user.created_at}") |
|
|
else: |
|
|
print("No users found in Supabase.") |
|
|
except Exception as e: |
|
|
print(f"Failed to list Supabase users: {e}") |
|
|
print("Please ensure your Supabase Anon Key has sufficient permissions or use a service role key if necessary.") |
|
|
|
|
|
async def main(): |
|
|
await test_db_connection() |
|
|
|
|
|
|
|
|
test_email = "test@example.com" |
|
|
test_password = "testpassword" |
|
|
await test_user_registration(test_email, test_password) |
|
|
|
|
|
await list_supabase_users() |
|
|
|
|
|
if __name__ == "__main__": |
|
|
import asyncio |
|
|
asyncio.run(main()) |
|
|
|