Add user invitation test (#4161)

backend/tests/integration/common_utils/managers/user.py

@@ -313,3 +313,29 @@ class UserManager:
         )
         response.raise_for_status()
         return UserInfo(**response.json())
+
+    @staticmethod
+    def invite_users(
+        user_performing_action: DATestUser,
+        emails: list[str],
+    ) -> int:
+        response = requests.put(
+            url=f"{API_SERVER_URL}/manage/admin/users",
+            json={"emails": emails},
+            headers=user_performing_action.headers,
+        )
+        response.raise_for_status()
+        return response.json()
+
+    @staticmethod
+    def remove_invited_user(
+        user_performing_action: DATestUser,
+        user_email: str,
+    ) -> int:
+        response = requests.patch(
+            url=f"{API_SERVER_URL}/manage/admin/remove-invited-user",
+            json={"user_email": user_email},
+            headers=user_performing_action.headers,
+        )
+        response.raise_for_status()
+        return response.json()

backend/tests/integration/common_utils/reset.py

@@ -22,6 +22,7 @@ from onyx.document_index.document_index_utils import get_multipass_config
 from onyx.document_index.vespa.index import DOCUMENT_ID_ENDPOINT
 from onyx.document_index.vespa.index import VespaIndex
 from onyx.indexing.models import IndexingSetting
+from onyx.redis.redis_pool import get_redis_client
 from onyx.setup import setup_postgres
 from onyx.setup import setup_vespa
 from onyx.utils.logger import setup_logger
@@ -237,6 +238,12 @@ def reset_vespa() -> None:
             time.sleep(5)
 
 
+def reset_redis() -> None:
+    """Reset the Redis database."""
+    redis_client = get_redis_client()
+    redis_client.flushall()
+
+
 def reset_postgres_multitenant() -> None:
     """Reset the Postgres database for all tenants in a multitenant setup."""
 
@@ -341,6 +348,8 @@ def reset_all() -> None:
     reset_postgres()
     logger.info("Resetting Vespa...")
     reset_vespa()
+    logger.info("Resetting Redis...")
+    reset_redis()
 
 
 def reset_all_multitenant() -> None:

backend/tests/integration/tests/auth/test_user_invitation.py

@@ -0,0 +1,38 @@
+import pytest
+from requests import HTTPError
+
+from onyx.auth.schemas import UserRole
+from tests.integration.common_utils.managers.user import UserManager
+from tests.integration.common_utils.test_models import DATestUser
+
+
+def test_inviting_users_flow(reset: None) -> None:
+    """
+    Test that verifies the functionality around inviting users:
+      1. Creating an admin user
+      2. Admin inviting a new user
+      3. Invited user successfully signing in
+      4. Non-invited user attempting to sign in (should result in an error)
+    """
+    # 1) Create an admin user (the first user created is automatically admin)
+    admin_user: DATestUser = UserManager.create(name="admin_user")
+    assert admin_user is not None
+    assert UserManager.is_role(admin_user, UserRole.ADMIN)
+
+    # 2) Admin invites a new user
+    invited_email = "invited_user@test.com"
+    invite_response = UserManager.invite_users(admin_user, [invited_email])
+
+    assert invite_response == 1
+
+    # 3) The invited user successfully registers/logs in
+    invited_user: DATestUser = UserManager.create(
+        name="invited_user", email=invited_email
+    )
+    assert invited_user is not None
+    assert invited_user.email == invited_email
+    assert UserManager.is_role(invited_user, UserRole.BASIC)
+
+    # 4) A non-invited user attempts to sign in/register (should fail)
+    with pytest.raises(HTTPError):
+        UserManager.create(name="uninvited_user", email="uninvited_user@test.com")