qzl
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
import argparse
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from v1.auth.gateway import SupabaseAuthGateway
|
|
from v1.auth.schemas import OtpSendRequest, EmailSessionCreateRequest
|
|
|
|
|
|
async def send_otp(email: str) -> None:
|
|
gateway = SupabaseAuthGateway()
|
|
print(f"Sending OTP email to: {email}")
|
|
print("-" * 50)
|
|
|
|
try:
|
|
request = OtpSendRequest(email=email)
|
|
await gateway.send_otp(request)
|
|
print("SUCCESS: OTP email sent!")
|
|
print("Please check your inbox for the 6-digit verification code.")
|
|
except Exception as e:
|
|
print(f"FAILED: {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
|
|
async def verify_otp(email: str, token: str) -> None:
|
|
gateway = SupabaseAuthGateway()
|
|
print(f"Verifying OTP for: {email}")
|
|
print(f"Token: {token}")
|
|
print("-" * 50)
|
|
|
|
try:
|
|
request = EmailSessionCreateRequest(email=email, token=token)
|
|
response = await gateway.create_email_session(request)
|
|
print("SUCCESS: Login successful!")
|
|
print(f"User ID: {response.user.id}")
|
|
print(f"Email: {response.user.email}")
|
|
print(f"Access Token: {response.access_token[:50]}...")
|
|
print(f"Expires In: {response.expires_in}s")
|
|
except Exception as e:
|
|
print(f"FAILED: {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Test Supabase OTP authentication")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
send_parser = subparsers.add_parser("send", help="Send OTP email")
|
|
send_parser.add_argument("email", help="Email address")
|
|
|
|
verify_parser = subparsers.add_parser("verify", help="Verify OTP code")
|
|
verify_parser.add_argument("email", help="Email address")
|
|
verify_parser.add_argument("token", help="6-digit OTP code")
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
if args.command == "send":
|
|
asyncio.run(send_otp(args.email))
|
|
elif args.command == "verify":
|
|
asyncio.run(verify_otp(args.email, args.token))
|
|
return 0
|
|
except Exception:
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|