Merge pull request #460 from supertokens/test/email-link

test: adds test for `create_reset_password_link`

Author: rishabhpoddar

tests/emailpassword/test_passwordreset.py

@@ -14,14 +14,20 @@
 import asyncio
 import json
 from typing import Any, Dict, Union
+from urllib.parse import urlparse
 
 from fastapi import FastAPI
 from fastapi.requests import Request
 from fastapi.testclient import TestClient
-from pytest import fixture, mark
+from pytest import fixture, mark, raises
 from supertokens_python import InputAppInfo, SupertokensConfig, init
+from supertokens_python.exceptions import GeneralError
 from supertokens_python.framework.fastapi import get_middleware
 from supertokens_python.recipe import emailpassword, session
+from supertokens_python.recipe.emailpassword.asyncio import create_reset_password_link
+from supertokens_python.recipe.emailpassword.interfaces import (
+    CreateResetPasswordLinkUnknownUserIdError,
+)
 from supertokens_python.recipe.session import SessionContainer
 from supertokens_python.recipe.session.asyncio import (
     create_new_session,
@@ -124,6 +130,7 @@ async def test_that_generated_password_link_is_correct(
     reset_url = None
     token_info: Union[None, str] = None
     rid_info: Union[None, str] = None
+    tenant_info: Union[None, str] = None
 
     class CustomEmailService(
         emailpassword.EmailDeliveryInterface[emailpassword.EmailTemplateVars]
@@ -133,11 +140,12 @@ async def send_email(
             template_vars: emailpassword.EmailTemplateVars,
             user_context: Dict[str, Any],
         ) -> None:
-            nonlocal reset_url, token_info, rid_info
+            nonlocal reset_url, token_info, rid_info, tenant_info
             password_reset_url_with_token = template_vars.password_reset_link
             reset_url = password_reset_url_with_token.split("?")[0]
             token_info = password_reset_url_with_token.split("?")[1].split("&")[0]
             rid_info = password_reset_url_with_token.split("?")[1].split("&")[1]
+            tenant_info = password_reset_url_with_token.split("?")[1].split("&")[2]
 
     init(
         supertokens_config=SupertokensConfig("http://localhost:3567"),
@@ -172,6 +180,7 @@ async def send_email(
     assert reset_url == "http://supertokens.io/auth/reset-password"
     assert token_info is not None and "token=" in token_info  # type: ignore pylint: disable=unsupported-membership-test
     assert rid_info is not None and "rid=emailpassword" in rid_info  # type: ignore pylint: disable=unsupported-membership-test
+    assert tenant_info is not None and "tenantId=public" in tenant_info  # type: ignore pylint: disable=unsupported-membership-test
 
 
 @mark.asyncio
@@ -258,6 +267,14 @@ async def send_email(
             token_info = (
                 password_reset_url_with_token.split("?")[1].split("&")[0].split("=")[1]
             )
+            assert (
+                password_reset_url_with_token.split("?")[1].split("&")[2].split("=")[1]
+                == "public"
+            )
+            assert (
+                password_reset_url_with_token.split("?")[1].split("&")[1].split("=")[1]
+                == "emailpassword"
+            )
 
     init(
         supertokens_config=SupertokensConfig("http://localhost:3567"),
@@ -339,3 +356,46 @@ async def send_email(
     assert dict_response["status"] == "OK"
     assert dict_response["user"]["id"] == user_info["id"]
     assert dict_response["user"]["email"] == user_info["email"]
+
+
+@mark.asyncio
+async def test_create_reset_password_link(
+    driver_config_client: TestClient,
+):
+    init(
+        supertokens_config=SupertokensConfig("http://localhost:3567"),
+        app_info=InputAppInfo(
+            app_name="SuperTokens Demo",
+            api_domain="http://api.supertokens.io",
+            website_domain="http://supertokens.io",
+            api_base_path="/auth",
+        ),
+        framework="fastapi",
+        recipe_list=[
+            emailpassword.init(),
+            session.init(get_token_transfer_method=lambda _, __, ___: "cookie"),
+        ],
+    )
+    start_st()
+
+    response_1 = sign_up_request(
+        driver_config_client, "[email protected]", "validpass123"
+    )
+    assert response_1.status_code == 200
+    dict_response = json.loads(response_1.text)
+    user_info = dict_response["user"]
+    assert dict_response["status"] == "OK"
+    link = await create_reset_password_link("public", user_info["id"])
+    url = urlparse(link.link)  # type: ignore
+    queries = url.query.strip("&").split("&")
+    assert url.path == "/auth/reset-password"
+    assert "token=" in queries[0]
+    assert "tenantId=public" in queries
+    assert "rid=emailpassword" in queries
+
+    link = await create_reset_password_link("public", "invalidUserId")
+    assert isinstance(link, CreateResetPasswordLinkUnknownUserIdError)
+
+    with raises(GeneralError) as err:
+        await create_reset_password_link("invalidTenantId", user_info["id"])
+    assert "status code: 400" in str(err.value)

tests/thirdpartyemailpassword/test_email_delivery.py

@@ -20,9 +20,11 @@
 from fastapi import FastAPI
 from fastapi.requests import Request
 from fastapi.testclient import TestClient
-from pytest import fixture, mark
+from pytest import fixture, mark, raises
+from urllib.parse import urlparse
 
 from supertokens_python import InputAppInfo, SupertokensConfig, init
+from supertokens_python.exceptions import GeneralError
 from supertokens_python.framework.fastapi import get_middleware
 from supertokens_python.ingredients.emaildelivery import EmailDeliveryInterface
 from supertokens_python.ingredients.emaildelivery.types import (
@@ -37,6 +39,15 @@
     session,
     thirdpartyemailpassword,
 )
+from supertokens_python.recipe.thirdpartyemailpassword.asyncio import (
+    create_reset_password_link,
+    send_reset_password_email,
+)
+from supertokens_python.recipe.thirdpartyemailpassword.interfaces import (
+    CreateResetPasswordLinkUnknownUserIdError,
+    SendResetPasswordEmailEmailOkResult,
+    SendResetPasswordEmailUnknownUserIdError,
+)
 from supertokens_python.recipe.emailverification.emaildelivery.services import (
     SMTPService as EVSMTPService,
 )
@@ -1036,3 +1047,139 @@ async def send_email(
 
     assert email == "[email protected]"
     assert email_verify_url != ""
+
+
+@mark.asyncio
+async def test_create_reset_password_link(
+    driver_config_client: TestClient,
+):
+    init(
+        supertokens_config=SupertokensConfig("http://localhost:3567"),
+        app_info=InputAppInfo(
+            app_name="SuperTokens Demo",
+            api_domain="http://api.supertokens.io",
+            website_domain="http://supertokens.io",
+            api_base_path="/auth",
+        ),
+        framework="fastapi",
+        recipe_list=[
+            thirdpartyemailpassword.init(),
+            session.init(get_token_transfer_method=lambda _, __, ___: "cookie"),
+        ],
+    )
+    start_st()
+
+    response_1 = sign_up_request(
+        driver_config_client, "[email protected]", "validpass123"
+    )
+    assert response_1.status_code == 200
+    dict_response = json.loads(response_1.text)
+    user_info = dict_response["user"]
+    assert dict_response["status"] == "OK"
+    link = await create_reset_password_link("public", user_info["id"])
+    url = urlparse(link.link)  # type: ignore
+    queries = url.query.strip("&").split("&")
+    assert url.path == "/auth/reset-password"
+    assert "tenantId=public" in queries
+    assert "rid=thirdpartyemailpassword" in queries
+
+    link = await create_reset_password_link("public", "invalidUserId")
+    assert isinstance(link, CreateResetPasswordLinkUnknownUserIdError)
+
+    with raises(GeneralError) as err:
+        await create_reset_password_link("invalidTenantId", user_info["id"])
+    assert "status code: 400" in str(err.value)
+
+
+@mark.asyncio
+async def test_send_reset_password_email(
+    driver_config_client: TestClient,
+):
+
+    tenant_info, token_info, rid_info, reset_url = "", "", "", ""
+
+    class CustomEmailDeliveryService(
+        thirdpartyemailpassword.EmailDeliveryInterface[PasswordResetEmailTemplateVars]
+    ):
+        async def send_email(
+            self,
+            template_vars: PasswordResetEmailTemplateVars,
+            user_context: Dict[str, Any],
+        ):
+            nonlocal reset_url, token_info, rid_info, tenant_info
+            password_reset_url = template_vars.password_reset_link
+            reset_url = password_reset_url.split("?")[0]
+            token_info = password_reset_url.split("?")[1].split("&")[0]
+            rid_info = password_reset_url.split("?")[1].split("&")[1]
+            tenant_info = password_reset_url.split("?")[1].split("&")[2]
+
+    init(
+        supertokens_config=SupertokensConfig("http://localhost:3567"),
+        app_info=InputAppInfo(
+            app_name="SuperTokens Demo",
+            api_domain="http://api.supertokens.io",
+            website_domain="http://supertokens.io",
+            api_base_path="/auth",
+        ),
+        framework="fastapi",
+        recipe_list=[
+            thirdpartyemailpassword.init(
+                email_delivery=thirdpartyemailpassword.EmailDeliveryConfig(
+                    CustomEmailDeliveryService()
+                )
+            ),
+            session.init(get_token_transfer_method=lambda _, __, ___: "cookie"),
+        ],
+    )
+    start_st()
+
+    response_1 = sign_up_request(
+        driver_config_client, "[email protected]", "validpass123"
+    )
+    assert response_1.status_code == 200
+    dict_response = json.loads(response_1.text)
+    user_info = dict_response["user"]
+    assert dict_response["status"] == "OK"
+    resp = await send_reset_password_email("public", user_info["id"])
+    assert isinstance(resp, SendResetPasswordEmailEmailOkResult)
+
+    assert reset_url == "http://supertokens.io/auth/reset-password"
+    assert token_info is not None and "token=" in token_info
+    assert rid_info is not None and "rid=thirdpartyemailpassword" in rid_info
+    assert tenant_info is not None and "tenantId=public" in tenant_info
+
+
+@mark.asyncio
+async def test_send_reset_password_email_invalid_input(
+    driver_config_client: TestClient,
+):
+
+    init(
+        supertokens_config=SupertokensConfig("http://localhost:3567"),
+        app_info=InputAppInfo(
+            app_name="SuperTokens Demo",
+            api_domain="http://api.supertokens.io",
+            website_domain="http://supertokens.io",
+            api_base_path="/auth",
+        ),
+        framework="fastapi",
+        recipe_list=[
+            thirdpartyemailpassword.init(),
+            session.init(get_token_transfer_method=lambda _, __, ___: "cookie"),
+        ],
+    )
+    start_st()
+
+    response_1 = sign_up_request(
+        driver_config_client, "[email protected]", "validpass123"
+    )
+    assert response_1.status_code == 200
+    dict_response = json.loads(response_1.text)
+    user_info = dict_response["user"]
+
+    link = await send_reset_password_email("public", "invalidUserId")
+    assert isinstance(link, SendResetPasswordEmailUnknownUserIdError)
+
+    with raises(GeneralError) as err:
+        await send_reset_password_email("invalidTenantId", user_info["id"])
+    assert "status code: 400" in str(err.value)