chore: gen sync files

gotrue/_sync/gotrue_client.py

@@ -352,8 +352,9 @@ def get_session(self) -> Union[Session, None]:
             current_session = self._in_memory_session
         if not current_session:
             return None
+        time_now = round(time())
         has_expired = (
-            current_session.expires_at <= time()
+            current_session.expires_at <= time_now + EXPIRY_MARGIN
             if current_session.expires_at
             else False
         )

tests/_sync/clients.py

@@ -0,0 +1,128 @@
+from jwt import encode
+
+from gotrue import SyncGoTrueAdminAPI, SyncGoTrueClient
+
+SIGNUP_ENABLED_AUTO_CONFIRM_OFF_PORT = 9999
+SIGNUP_ENABLED_AUTO_CONFIRM_ON_PORT = 9998
+SIGNUP_DISABLED_AUTO_CONFIRM_OFF_PORT = 9997
+
+GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_OFF = (
+    f"http://localhost:{SIGNUP_ENABLED_AUTO_CONFIRM_OFF_PORT}"
+)
+GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON = (
+    f"http://localhost:{SIGNUP_ENABLED_AUTO_CONFIRM_ON_PORT}"
+)
+GOTRUE_URL_SIGNUP_DISABLED_AUTO_CONFIRM_OFF = (
+    f"http://localhost:{SIGNUP_DISABLED_AUTO_CONFIRM_OFF_PORT}"
+)
+
+GOTRUE_JWT_SECRET = "37c304f8-51aa-419a-a1af-06154e63707a"
+
+AUTH_ADMIN_JWT = encode(
+    {
+        "sub": "1234567890",
+        "role": "supabase_admin",
+    },
+    GOTRUE_JWT_SECRET,
+)
+
+
+def auth_client():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        auto_refresh_token=False,
+        persist_session=True,
+    )
+
+
+def auth_client_with_session():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        auto_refresh_token=False,
+        persist_session=False,
+    )
+
+
+def auth_subscription_client():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        auto_refresh_token=False,
+        persist_session=True,
+    )
+
+
+def client_api_auto_confirm_enabled_client():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        auto_refresh_token=False,
+        persist_session=True,
+    )
+
+
+def client_api_auto_confirm_off_signups_enabled_client():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_OFF,
+        auto_refresh_token=False,
+        persist_session=True,
+    )
+
+
+def client_api_auto_confirm_disabled_client():
+    return SyncGoTrueClient(
+        url=GOTRUE_URL_SIGNUP_DISABLED_AUTO_CONFIRM_OFF,
+        auto_refresh_token=False,
+        persist_session=True,
+    )
+
+
+def auth_admin_api_auto_confirm_enabled_client():
+    return SyncGoTrueAdminAPI(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        headers={
+            "Authorization": f"Bearer {AUTH_ADMIN_JWT}",
+        },
+    )
+
+
+def auth_admin_api_auto_confirm_disabled_client():
+    return SyncGoTrueAdminAPI(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_OFF,
+        headers={
+            "Authorization": f"Bearer {AUTH_ADMIN_JWT}",
+        },
+    )
+
+
+SERVICE_ROLE_JWT = encode(
+    {
+        "role": "service_role",
+    },
+    GOTRUE_JWT_SECRET,
+)
+
+
+def service_role_api_client():
+    return SyncGoTrueAdminAPI(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_ON,
+        headers={
+            "Authorization": f"Bearer {SERVICE_ROLE_JWT}",
+        },
+    )
+
+
+def service_role_api_client_with_sms():
+    return SyncGoTrueAdminAPI(
+        url=GOTRUE_URL_SIGNUP_ENABLED_AUTO_CONFIRM_OFF,
+        headers={
+            "Authorization": f"Bearer {SERVICE_ROLE_JWT}",
+        },
+    )
+
+
+def service_role_api_client_no_sms():
+    return SyncGoTrueAdminAPI(
+        url=GOTRUE_URL_SIGNUP_DISABLED_AUTO_CONFIRM_OFF,
+        headers={
+            "Authorization": f"Bearer {SERVICE_ROLE_JWT}",
+        },
+    )

tests/_sync/test_gotrue_admin_api.py

@@ -0,0 +1,61 @@
+from .clients import service_role_api_client
+from .utils import (
+    create_new_user_with_email,
+    mock_app_metadata,
+    mock_user_credentials,
+    mock_user_metadata,
+)
+
+
+def test_create_user_should_create_a_new_user():
+    credentials = mock_user_credentials()
+    response = create_new_user_with_email(email=credentials.get("email"))
+    assert response.email == credentials.get("email")
+
+
+def test_create_user_with_user_metadata():
+    user_metadata = mock_user_metadata()
+    credentials = mock_user_credentials()
+    response = service_role_api_client().create_user(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+            "user_metadata": user_metadata,
+        }
+    )
+    assert response.user.email == credentials.get("email")
+    assert response.user.user_metadata == user_metadata
+    assert "profile_image" in response.user.user_metadata
+
+
+def test_create_user_with_app_metadata():
+    app_metadata = mock_app_metadata()
+    credentials = mock_user_credentials()
+    response = service_role_api_client().create_user(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+            "app_metadata": app_metadata,
+        }
+    )
+    assert response.user.email == credentials.get("email")
+    assert "provider" in response.user.app_metadata
+    assert "providers" in response.user.app_metadata
+
+
+def test_create_user_with_user_and_app_metadata():
+    user_metadata = mock_user_metadata()
+    app_metadata = mock_app_metadata()
+    credentials = mock_user_credentials()
+    response = service_role_api_client().create_user(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+            "user_metadata": user_metadata,
+            "app_metadata": app_metadata,
+        }
+    )
+    assert response.user.email == credentials.get("email")
+    assert "profile_image" in response.user.user_metadata
+    assert "provider" in response.user.app_metadata
+    assert "providers" in response.user.app_metadata

tests/_sync/test_utils.py

@@ -0,0 +1,38 @@
+from time import time
+
+from .utils import (
+    create_new_user_with_email,
+    mock_app_metadata,
+    mock_user_credentials,
+    mock_user_metadata,
+)
+
+
+def test_mock_user_credentials_has_email():
+    credentials = mock_user_credentials()
+    assert credentials.get("email")
+    assert credentials.get("password")
+
+
+def test_mock_user_credentials_has_phone():
+    credentials = mock_user_credentials()
+    assert credentials.get("phone")
+    assert credentials.get("password")
+
+
+def test_create_new_user_with_email():
+    email = f"user+{int(time())}@example.com"
+    user = create_new_user_with_email(email=email)
+    assert user.email == email
+
+
+def test_mock_user_metadata():
+    user_metadata = mock_user_metadata()
+    assert user_metadata
+    assert user_metadata.get("profile_image")
+
+
+def test_mock_app_metadata():
+    app_metadata = mock_app_metadata()
+    assert app_metadata
+    assert app_metadata.get("roles")

tests/_sync/utils.py

@@ -0,0 +1,82 @@
+from random import random
+from time import time
+from typing import Union
+
+from faker import Faker
+from jwt import encode
+from typing_extensions import NotRequired, TypedDict
+
+from gotrue.types import User
+
+from .clients import GOTRUE_JWT_SECRET, service_role_api_client
+
+
+def mock_access_token() -> str:
+    return encode(
+        {
+            "sub": "1234567890",
+            "role": "anon_key",
+        },
+        GOTRUE_JWT_SECRET,
+    )
+
+
+class OptionalCredentials(TypedDict):
+    email: NotRequired[Union[str, None]]
+    phone: NotRequired[Union[str, None]]
+    password: NotRequired[Union[str, None]]
+
+
+class Credentials(TypedDict):
+    email: str
+    phone: str
+    password: str
+
+
+def mock_user_credentials(
+    options: OptionalCredentials = {},
+) -> Credentials:
+    fake = Faker()
+    rand_numbers = str(time())
+    return {
+        "email": options.get("email") or fake.email(),
+        "phone": options.get("phone") or f"1{rand_numbers[-11:]}",
+        "password": options.get("password") or fake.password(),
+    }
+
+
+def mock_verification_otp() -> str:
+    return str(int(100000 + random() * 900000))
+
+
+def mock_user_metadata():
+    fake = Faker()
+    return {
+        "profile_image": fake.url(),
+    }
+
+
+def mock_app_metadata():
+    return {
+        "roles": ["editor", "publisher"],
+    }
+
+
+def create_new_user_with_email(
+    *,
+    email: Union[str, None] = None,
+    password: Union[str, None] = None,
+) -> User:
+    credentials = mock_user_credentials(
+        {
+            "email": email,
+            "password": password,
+        }
+    )
+    response = service_role_api_client().create_user(
+        {
+            "email": credentials["email"],
+            "password": credentials["password"],
+        }
+    )
+    return response.user