[OND211-2329]: Updated team API's and added create group tests.
This commit is contained in:
Hetavi Shah
parent
5af04fdf45
commit
3e7d57b208
5 changed files with 870 additions and 19 deletions
|
|
@ -17,7 +17,7 @@
|
|||
import base64
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
|
@ -81,7 +81,13 @@ def encrypt_password(password: str) -> str:
|
|||
return base64.b64encode(encrypted_password).decode()
|
||||
|
||||
|
||||
def register():
|
||||
def register() -> Optional[str]:
|
||||
"""Register the test user.
|
||||
|
||||
Returns:
|
||||
str: Authorization token if registration succeeded and user was logged in,
|
||||
None if user already exists
|
||||
"""
|
||||
url: str = HOST_ADDRESS + f"/{VERSION}/user/register"
|
||||
name: str = "qa"
|
||||
# Encrypt the plain password "123" before sending
|
||||
|
|
@ -89,9 +95,21 @@ def register():
|
|||
encrypted_password: str = encrypt_password(plain_password)
|
||||
register_data = {"email": EMAIL, "nickname": name, "password": encrypted_password}
|
||||
res: requests.Response = requests.post(url=url, json=register_data)
|
||||
res: Dict[str, Any] = res.json()
|
||||
if res.get("code") != 0 and "has already registered" not in res.get("message"):
|
||||
raise Exception(res.get("message"))
|
||||
res_json: Dict[str, Any] = res.json()
|
||||
if res_json.get("code") != 0 and "has already registered" not in res_json.get("message"):
|
||||
print(f"Registration failed with code {res_json.get('code')}: {res_json.get('message')}")
|
||||
raise Exception(res_json.get("message"))
|
||||
elif res_json.get("code") == 0:
|
||||
print(f"Registration successful for {EMAIL}")
|
||||
# Registration endpoint logs user in and returns auth token
|
||||
auth_token: str = res.headers.get("Authorization", "")
|
||||
if auth_token:
|
||||
print(f"Received auth token from registration")
|
||||
return auth_token
|
||||
else:
|
||||
print(f"Warning: No auth token in registration response")
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def login():
|
||||
|
|
@ -108,6 +126,102 @@ def login():
|
|||
return auth
|
||||
|
||||
|
||||
def delete_user_from_db(email: str) -> bool:
|
||||
"""Delete a user directly from the database using SQL.
|
||||
|
||||
This is a helper function for cleanup when a user exists with wrong password.
|
||||
Uses direct SQL to avoid import conflicts with test helper modules.
|
||||
|
||||
Args:
|
||||
email: Email of the user to delete
|
||||
|
||||
Returns:
|
||||
bool: True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
import subprocess
|
||||
import sys
|
||||
import os
|
||||
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.abspath(os.path.join(current_dir, "..", ".."))
|
||||
|
||||
# Create a temporary Python script to hard delete the user
|
||||
delete_script = f"""
|
||||
import sys
|
||||
sys.path.insert(0, '{project_root}')
|
||||
|
||||
# Remove test directories from path to avoid conflicts
|
||||
test_paths = [p for p in sys.path if 'test/testcases' in p or 'testcases' in p]
|
||||
for p in test_paths:
|
||||
if p in sys.path:
|
||||
sys.path.remove(p)
|
||||
|
||||
try:
|
||||
from api.db.db_models import DB, User, Tenant, UserTenant, File
|
||||
|
||||
users = list(User.select().where(User.email == '{email}'))
|
||||
if users:
|
||||
with DB.atomic():
|
||||
for user in users:
|
||||
user_id = user.id
|
||||
# Hard delete related records and user
|
||||
try:
|
||||
# Delete user-tenant relationships
|
||||
UserTenant.delete().where(UserTenant.user_id == user_id).execute()
|
||||
UserTenant.delete().where(UserTenant.tenant_id == user_id).execute()
|
||||
|
||||
# Delete files owned by user
|
||||
File.delete().where(File.created_by == user_id).execute()
|
||||
File.delete().where(File.tenant_id == user_id).execute()
|
||||
|
||||
# Delete tenant
|
||||
Tenant.delete().where(Tenant.id == user_id).execute()
|
||||
|
||||
# Finally delete user
|
||||
User.delete().where(User.id == user_id).execute()
|
||||
except Exception as e:
|
||||
print(f"Warning during cleanup: {{e}}")
|
||||
print(f"DELETED_USER:{email}")
|
||||
else:
|
||||
print(f"USER_NOT_FOUND:{email}")
|
||||
except Exception as e:
|
||||
print(f"ERROR:{{e}}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
"""
|
||||
|
||||
# Run the delete script in a subprocess to avoid import conflicts
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", delete_script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
output = result.stdout + result.stderr
|
||||
|
||||
if "DELETED_USER:" in output:
|
||||
print(f"Successfully deleted user {email} from database")
|
||||
return True
|
||||
elif "USER_NOT_FOUND:" in output:
|
||||
print(f"User {email} not found in database")
|
||||
return False
|
||||
else:
|
||||
print(f"Failed to delete user from database")
|
||||
if output:
|
||||
print(f"Output: {output}")
|
||||
return False
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
print(f"Timeout while trying to delete user from database")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"Failed to delete user from database: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def login_as_user(email: str, password: str) -> RAGFlowWebApiAuth:
|
||||
"""Login as a user and return authentication object.
|
||||
|
||||
|
|
@ -140,21 +254,88 @@ def login_as_user(email: str, password: str) -> RAGFlowWebApiAuth:
|
|||
|
||||
@pytest.fixture(scope="session")
|
||||
def auth():
|
||||
try:
|
||||
register()
|
||||
except Exception as e:
|
||||
print(e)
|
||||
"""Session fixture to authenticate test user.
|
||||
|
||||
This fixture tries to login with the test user. If login fails because
|
||||
the user doesn't exist, it registers the user first. If the user exists
|
||||
with a different password, it provides instructions to fix the issue.
|
||||
|
||||
Returns:
|
||||
str: Authentication token
|
||||
|
||||
Raises:
|
||||
Exception: If authentication fails
|
||||
"""
|
||||
# First, try to login (user might already exist with correct password)
|
||||
try:
|
||||
auth: str = login()
|
||||
print(f"Successfully logged in as {EMAIL}")
|
||||
return auth
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
if "Email and password do not match" in error_msg:
|
||||
except Exception as login_error:
|
||||
login_error_msg = str(login_error)
|
||||
|
||||
# If user doesn't exist, try to register
|
||||
if "is not registered" in login_error_msg:
|
||||
print(f"User {EMAIL} not found, attempting to register...")
|
||||
try:
|
||||
auth_token: Optional[str] = register()
|
||||
if auth_token:
|
||||
print(f"Successfully registered and logged in as {EMAIL}")
|
||||
return auth_token
|
||||
else:
|
||||
# Try login if register didn't return auth token
|
||||
auth: str = login()
|
||||
print(f"Successfully registered and logged in as {EMAIL}")
|
||||
return auth
|
||||
except Exception as register_error:
|
||||
raise Exception(
|
||||
f"Failed to register user {EMAIL}: {register_error}"
|
||||
) from register_error
|
||||
|
||||
# If user exists but password doesn't match
|
||||
elif "Email and password do not match" in login_error_msg:
|
||||
print(f"User {EMAIL} exists but password doesn't match. Attempting to delete and recreate...")
|
||||
|
||||
# Try to delete the user from database directly
|
||||
if delete_user_from_db(EMAIL):
|
||||
# Delay to ensure deletion is committed to database
|
||||
time.sleep(1.0)
|
||||
|
||||
# Now try to register and login
|
||||
try:
|
||||
print(f"Attempting to register user {EMAIL}...")
|
||||
auth_token: Optional[str] = register()
|
||||
if auth_token:
|
||||
print(f"Successfully recreated user {EMAIL} with correct password")
|
||||
return auth_token
|
||||
else:
|
||||
# Try login if register didn't return auth token
|
||||
print(f"Registration completed, now attempting login...")
|
||||
auth: str = login()
|
||||
print(f"Successfully recreated user {EMAIL} with correct password")
|
||||
return auth
|
||||
except Exception as recreate_error:
|
||||
recreate_error_msg = str(recreate_error)
|
||||
print(f"Recreation failed: {recreate_error_msg}")
|
||||
raise Exception(
|
||||
f"Failed to recreate user after deletion: {recreate_error_msg}"
|
||||
) from recreate_error
|
||||
else:
|
||||
# If database deletion failed, provide instructions
|
||||
raise Exception(
|
||||
f"Login failed: User {EMAIL} exists but password doesn't match.\n"
|
||||
f"Automatic cleanup failed. To fix this issue:\n"
|
||||
f"1. Manually delete the user from the database, OR\n"
|
||||
f"2. Reset the password in the database to '123', OR\n"
|
||||
f"3. Update EMAIL in configs.py to use a different test user\n"
|
||||
f"Original error: {login_error_msg}"
|
||||
) from login_error
|
||||
|
||||
# Other login errors
|
||||
else:
|
||||
raise Exception(
|
||||
f"Login failed: User {EMAIL} exists but password doesn't match. "
|
||||
f"Please ensure the user has the correct password or delete the user first."
|
||||
) from e
|
||||
raise
|
||||
f"Login failed with unexpected error: {login_error_msg}"
|
||||
) from login_error
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
|
|
|
|||
|
|
@ -477,8 +477,36 @@ def remove_users_from_team(
|
|||
return res.json()
|
||||
|
||||
|
||||
def accept_team_invitation(
|
||||
auth: Union[AuthBase, str, None],
|
||||
tenant_id: str,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Accept a team invitation.
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
tenant_id: The tenant/team ID to accept invitation for.
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing the acceptance result.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{TEAM_API_URL}/update-request/{tenant_id}"
|
||||
payload: Dict[str, bool] = {"accept": True}
|
||||
res: requests.Response = requests.put(
|
||||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
# DEPARTMENT MANAGEMENT
|
||||
DEPARTMENT_API_URL: str = f"/{VERSION}/department"
|
||||
GROUP_API_URL: str = f"/{VERSION}/group"
|
||||
|
||||
|
||||
def create_department(
|
||||
|
|
@ -641,3 +669,30 @@ def list_department_members(
|
|||
url=url, headers=headers, auth=auth
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
# GROUP MANAGEMENT
|
||||
def create_group(
|
||||
auth: Union[AuthBase, str, None],
|
||||
payload: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a new group.
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
payload: Optional JSON payload containing group data (e.g., name, tenant_id, description).
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing the created group data.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{GROUP_API_URL}/create"
|
||||
res: requests.Response = requests.post(
|
||||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
|
|
|||
|
|
@ -15,12 +15,14 @@
|
|||
#
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from common import (
|
||||
accept_team_invitation,
|
||||
add_department_members,
|
||||
add_users_to_team,
|
||||
create_department,
|
||||
|
|
@ -187,6 +189,20 @@ class TestAddMembers:
|
|||
assert set(res["data"]["added"]) == set(user_ids)
|
||||
assert len(res["data"]["failed"]) == 0
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_missing_request_body(
|
||||
self,
|
||||
web_api_auth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test adding members without request body."""
|
||||
res: dict[str, Any] = add_department_members(
|
||||
web_api_auth, test_department["id"], None
|
||||
)
|
||||
|
||||
assert res["code"] == 101
|
||||
assert "required" in res["message"].lower() or "body" in res["message"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_missing_user_ids(
|
||||
self,
|
||||
|
|
@ -200,7 +216,7 @@ class TestAddMembers:
|
|||
)
|
||||
|
||||
assert res["code"] == 101
|
||||
assert "user_ids" in res["message"].lower() or "required" in res[
|
||||
assert "user_ids" in res["message"].lower() or "non-empty array" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
|
|
@ -412,6 +428,9 @@ class TestAddMembers:
|
|||
add_team_res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_team_payload)
|
||||
assert add_team_res["code"] == 0, "Failed to add user to team"
|
||||
|
||||
# Wait a bit for user creation and team addition to be committed
|
||||
time.sleep(0.3)
|
||||
|
||||
# Create another user to add to the department
|
||||
another_user_email: str = f"anotheruser_{uuid.uuid4().hex[:8]}@example.com"
|
||||
another_user_password: str = "TestPassword123!"
|
||||
|
|
@ -431,8 +450,23 @@ class TestAddMembers:
|
|||
)
|
||||
assert add_another_team_res["code"] == 0, "Failed to add another user to team"
|
||||
|
||||
# Wait a bit for team addition to be committed
|
||||
time.sleep(0.3)
|
||||
|
||||
# Login as the normal user (not admin/owner)
|
||||
normal_user_auth: RAGFlowWebApiAuth = login_as_user(normal_user_email, normal_user_password)
|
||||
# Users with INVITE role should still be able to login
|
||||
try:
|
||||
normal_user_auth: RAGFlowWebApiAuth = login_as_user(normal_user_email, normal_user_password)
|
||||
except Exception as e:
|
||||
pytest.skip(f"Failed to login as normal user: {e}")
|
||||
|
||||
# Accept the invitation to join the team
|
||||
accept_res: dict[str, Any] = accept_team_invitation(normal_user_auth, tenant_id)
|
||||
if accept_res["code"] != 0:
|
||||
pytest.skip(f"Failed to accept invitation: {accept_res.get('message', 'Unknown error')}")
|
||||
|
||||
# Wait a bit for invitation acceptance to be committed
|
||||
time.sleep(0.2)
|
||||
|
||||
# Try to add a member as the normal user (should fail)
|
||||
add_payload: dict[str, list[str]] = {"user_ids": [another_user_id]}
|
||||
|
|
|
|||
|
|
@ -15,12 +15,14 @@
|
|||
#
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from common import (
|
||||
accept_team_invitation,
|
||||
add_department_members,
|
||||
add_users_to_team,
|
||||
create_department,
|
||||
|
|
@ -337,6 +339,9 @@ class TestRemoveMember:
|
|||
add_team_res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_team_payload)
|
||||
assert add_team_res["code"] == 0, "Failed to add user to team"
|
||||
|
||||
# Wait a bit for user creation and team addition to be committed
|
||||
time.sleep(0.3)
|
||||
|
||||
# Add another user to the department (so we have someone to remove)
|
||||
another_user_email: str = f"anotheruser_{uuid.uuid4().hex[:8]}@example.com"
|
||||
another_user_password: str = "TestPassword123!"
|
||||
|
|
@ -356,6 +361,9 @@ class TestRemoveMember:
|
|||
)
|
||||
assert add_another_team_res["code"] == 0, "Failed to add another user to team"
|
||||
|
||||
# Wait a bit for team addition to be committed
|
||||
time.sleep(0.3)
|
||||
|
||||
# Add another user to the department (as owner/admin)
|
||||
add_dept_payload: dict[str, list[str]] = {"user_ids": [another_user_id]}
|
||||
add_dept_res: dict[str, Any] = add_department_members(
|
||||
|
|
@ -364,7 +372,19 @@ class TestRemoveMember:
|
|||
assert add_dept_res["code"] == 0, "Failed to add user to department"
|
||||
|
||||
# Login as the normal user (not admin/owner)
|
||||
normal_user_auth: RAGFlowWebApiAuth = login_as_user(normal_user_email, normal_user_password)
|
||||
# Users with INVITE role should still be able to login
|
||||
try:
|
||||
normal_user_auth: RAGFlowWebApiAuth = login_as_user(normal_user_email, normal_user_password)
|
||||
except Exception as e:
|
||||
pytest.skip(f"Failed to login as normal user: {e}")
|
||||
|
||||
# Accept the invitation to join the team
|
||||
accept_res: dict[str, Any] = accept_team_invitation(normal_user_auth, tenant_id)
|
||||
if accept_res["code"] != 0:
|
||||
pytest.skip(f"Failed to accept invitation: {accept_res.get('message', 'Unknown error')}")
|
||||
|
||||
# Wait a bit for invitation acceptance to be committed
|
||||
time.sleep(0.2)
|
||||
|
||||
# Try to remove a member as the normal user (should fail)
|
||||
remove_res: dict[str, Any] = remove_department_member(
|
||||
|
|
@ -398,3 +418,79 @@ class TestRemoveMember:
|
|||
assert len(add_res["data"]["added"]) == 1
|
||||
assert add_res["data"]["added"][0] == user_id
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_remove_multiple_members_sequentially(
|
||||
self,
|
||||
web_api_auth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
test_team: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing multiple members sequentially."""
|
||||
# Create multiple users
|
||||
users = []
|
||||
for i in range(3):
|
||||
email = f"testuser{i}_{uuid.uuid4().hex[:8]}@example.com"
|
||||
user_payload: dict[str, str] = {
|
||||
"email": email,
|
||||
"password": "TestPassword123!",
|
||||
"nickname": f"Test User {i}",
|
||||
}
|
||||
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
|
||||
if user_res["code"] == 0:
|
||||
users.append({"email": email, "id": user_res["data"]["id"]})
|
||||
|
||||
if len(users) < 2:
|
||||
pytest.skip("Need at least 2 test users")
|
||||
|
||||
# Add users to team
|
||||
for user in users:
|
||||
add_team_payload: dict[str, list[str]] = {"users": [user["email"]]}
|
||||
add_users_to_team(web_api_auth, test_team["id"], add_team_payload)
|
||||
|
||||
# Add users to department
|
||||
user_ids: list[str] = [user["id"] for user in users]
|
||||
add_dept_payload: dict[str, list[str]] = {"user_ids": user_ids}
|
||||
add_res: dict[str, Any] = add_department_members(
|
||||
web_api_auth, test_department["id"], add_dept_payload
|
||||
)
|
||||
assert add_res["code"] == 0
|
||||
assert len(add_res["data"]["added"]) == len(users)
|
||||
|
||||
# Remove all users sequentially
|
||||
for user in users:
|
||||
remove_res: dict[str, Any] = remove_department_member(
|
||||
web_api_auth, test_department["id"], user["id"]
|
||||
)
|
||||
assert remove_res["code"] == 0
|
||||
assert "removed" in remove_res["message"].lower() or "success" in remove_res["message"].lower()
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_remove_member_empty_string_user_id(
|
||||
self,
|
||||
web_api_auth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing a member with empty string user ID."""
|
||||
res: dict[str, Any] = remove_department_member(
|
||||
web_api_auth, test_department["id"], ""
|
||||
)
|
||||
|
||||
# Empty string user ID may return 100 (EXCEPTION_ERROR) or 102 (DATA_ERROR)
|
||||
assert res["code"] in [100, 102]
|
||||
assert "not a member" in res["message"].lower() or "not found" in res["message"].lower() or "error" in res["message"].lower()
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_remove_member_special_characters_user_id(
|
||||
self,
|
||||
web_api_auth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing a member with special characters in user ID."""
|
||||
invalid_user_id: str = "user@123_!@#$%"
|
||||
res: dict[str, Any] = remove_department_member(
|
||||
web_api_auth, test_department["id"], invalid_user_id
|
||||
)
|
||||
|
||||
assert res["code"] == 102
|
||||
assert "not a member" in res["message"].lower() or "not found" in res["message"].lower()
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,485 @@
|
|||
#
|
||||
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
#
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import os
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from Cryptodome.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5
|
||||
from Cryptodome.PublicKey import RSA
|
||||
|
||||
from common import create_group, create_team, create_user, login_as_user
|
||||
from configs import INVALID_API_TOKEN
|
||||
from libs.auth import RAGFlowWebApiAuth
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper Functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def encrypt_password(password: str) -> str:
|
||||
"""Encrypt password using RSA for API calls.
|
||||
|
||||
Args:
|
||||
password: Plain text password to encrypt.
|
||||
|
||||
Returns:
|
||||
Encrypted password as a base64-encoded string.
|
||||
"""
|
||||
current_dir: str = os.path.dirname(os.path.abspath(__file__))
|
||||
project_base: str = os.path.abspath(os.path.join(current_dir, "..", "..", "..", ".."))
|
||||
file_path: str = os.path.join(project_base, "conf", "public.pem")
|
||||
|
||||
with open(file_path, encoding="utf-8") as pem_file:
|
||||
rsa_key: RSA.RsaKey = RSA.import_key(pem_file.read(), passphrase="Welcome")
|
||||
|
||||
cipher: Cipher_pkcs1_v1_5.PKCS115_Cipher = Cipher_pkcs1_v1_5.new(rsa_key)
|
||||
password_base64: str = base64.b64encode(password.encode()).decode()
|
||||
encrypted_password: bytes = cipher.encrypt(password_base64.encode())
|
||||
return base64.b64encode(encrypted_password).decode()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test Classes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestAuthorization:
|
||||
"""Tests for authentication behavior during group creation."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("invalid_auth", "expected_code", "expected_message"),
|
||||
[
|
||||
# Endpoint now requires @login_required (JWT token auth)
|
||||
(None, 401, "Unauthorized"),
|
||||
(RAGFlowWebApiAuth(INVALID_API_TOKEN), 401, "Unauthorized"),
|
||||
],
|
||||
)
|
||||
def test_invalid_auth(
|
||||
self,
|
||||
invalid_auth: RAGFlowWebApiAuth | None,
|
||||
expected_code: int,
|
||||
expected_message: str,
|
||||
web_api_auth: RAGFlowWebApiAuth,
|
||||
) -> None:
|
||||
"""Test group creation with invalid or missing authentication."""
|
||||
# First create a team to use as tenant_id
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
if team_res["code"] != 0:
|
||||
pytest.skip("Team creation failed, skipping auth test")
|
||||
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Try to create group with invalid auth
|
||||
group_payload: dict[str, str] = {
|
||||
"name": "Test Group Auth",
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(invalid_auth, group_payload)
|
||||
assert res["code"] == expected_code, res
|
||||
if expected_message:
|
||||
assert expected_message in res["message"]
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestGroupCreate:
|
||||
"""Comprehensive tests for group creation API."""
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_with_name_and_tenant_id(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with name and tenant_id."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create group
|
||||
group_name: str = f"Test Group {uuid.uuid4().hex[:8]}"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 0, res
|
||||
assert "data" in res
|
||||
assert res["data"]["name"] == group_name
|
||||
assert res["data"]["tenant_id"] == tenant_id
|
||||
assert "id" in res["data"]
|
||||
assert "created successfully" in res["message"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_with_description(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with description."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create group with description
|
||||
group_name: str = f"Test Group {uuid.uuid4().hex[:8]}"
|
||||
description: str = "This is a test group description"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
"description": description,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 0, res
|
||||
assert res["data"]["name"] == group_name
|
||||
assert res["data"]["description"] == description
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_missing_name(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group without name."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Try to create group without name
|
||||
group_payload: dict[str, str] = {"tenant_id": tenant_id}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 101
|
||||
assert "name" in res["message"].lower() or "required" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_empty_name(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with empty name."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Try to create group with empty name
|
||||
group_payload: dict[str, str] = {"name": "", "tenant_id": tenant_id}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 101
|
||||
assert "name" in res["message"].lower() or "empty" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_missing_tenant_id(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group without tenant_id."""
|
||||
# Try to create group without tenant_id
|
||||
group_payload: dict[str, str] = {"name": "Test Group"}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 101
|
||||
assert "tenant_id" in res["message"].lower() or "required" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_invalid_tenant_id(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with non-existent tenant_id."""
|
||||
group_payload: dict[str, str] = {
|
||||
"name": "Test Group Invalid Tenant",
|
||||
"tenant_id": "non_existent_tenant_id_12345",
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
# Permission check happens before tenant existence check,
|
||||
# so invalid tenant_id results in permission error (108) not data error (102)
|
||||
assert res["code"] == 108
|
||||
assert "only team owners or admins" in res["message"].lower() or "permission" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_name_too_long(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with name exceeding 128 characters."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Try to create group with name too long
|
||||
# Note: The API validates name length (max 128), so it should return an error
|
||||
long_name: str = "A" * 129
|
||||
group_payload: dict[str, str] = {
|
||||
"name": long_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
# API validates length, so it should return an argument error
|
||||
assert res["code"] == 101
|
||||
assert "128" in res["message"] or "characters" in res["message"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_not_team_owner_or_admin(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group when user is not team owner or admin."""
|
||||
# Create a team with the main user (owner)
|
||||
team_name: str = f"Owner Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create a second user with encrypted password (now supported!)
|
||||
other_user_email: str = f"test_{uuid.uuid4().hex[:8]}@example.com"
|
||||
other_user_password: str = "test123"
|
||||
encrypted_password: str = encrypt_password(other_user_password)
|
||||
|
||||
user_payload: dict[str, str] = {
|
||||
"nickname": "Other User",
|
||||
"email": other_user_email,
|
||||
"password": encrypted_password, # Now works with encryption!
|
||||
}
|
||||
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
|
||||
assert user_res["code"] == 0, user_res
|
||||
|
||||
# Small delay to ensure user is fully created
|
||||
import time
|
||||
time.sleep(0.5)
|
||||
|
||||
# Login as the other user
|
||||
other_user_auth: RAGFlowWebApiAuth = login_as_user(
|
||||
other_user_email, other_user_password
|
||||
)
|
||||
|
||||
# Try to create a group in the owner's team as the other user
|
||||
group_name: str = f"Group {uuid.uuid4().hex[:8]}"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(other_user_auth, group_payload)
|
||||
|
||||
# Should fail - user is not the team owner or admin
|
||||
assert res["code"] != 0, (
|
||||
"Non-owner/non-admin should not be able to create groups in another user's team"
|
||||
)
|
||||
|
||||
# Verify it's a permission-related error
|
||||
# Common permission error codes: 108 (Permission denied), 403 (Forbidden), 104 (Permission Error), 102 (Authentication Error)
|
||||
assert res["code"] in [108, 403, 104, 102], (
|
||||
f"Expected permission error code (108, 403, 104, or 102), got: {res}"
|
||||
)
|
||||
|
||||
# Verify the error message indicates permission issue
|
||||
assert "owner" in res["message"].lower() or "admin" in res["message"].lower() or "permission" in res["message"].lower(), (
|
||||
f"Error message should indicate permission issue, got: {res['message']}"
|
||||
)
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_response_structure(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test that group creation returns the expected response structure."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create group
|
||||
group_name: str = f"Test Group Structure {uuid.uuid4().hex[:8]}"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 0
|
||||
assert "data" in res
|
||||
assert isinstance(res["data"], dict)
|
||||
assert "id" in res["data"]
|
||||
assert "name" in res["data"]
|
||||
assert "tenant_id" in res["data"]
|
||||
assert res["data"]["name"] == group_name
|
||||
assert res["data"]["tenant_id"] == tenant_id
|
||||
assert "message" in res
|
||||
assert "created successfully" in res["message"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_multiple_groups_same_team(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating multiple groups for the same team."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create first group
|
||||
group_name_1: str = f"Group 1 {uuid.uuid4().hex[:8]}"
|
||||
group_payload_1: dict[str, str] = {
|
||||
"name": group_name_1,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res1: dict[str, Any] = create_group(web_api_auth, group_payload_1)
|
||||
assert res1["code"] == 0, res1
|
||||
group_id_1: str = res1["data"]["id"]
|
||||
|
||||
# Create second group
|
||||
group_name_2: str = f"Group 2 {uuid.uuid4().hex[:8]}"
|
||||
group_payload_2: dict[str, str] = {
|
||||
"name": group_name_2,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res2: dict[str, Any] = create_group(web_api_auth, group_payload_2)
|
||||
assert res2["code"] == 0, res2
|
||||
group_id_2: str = res2["data"]["id"]
|
||||
|
||||
# Verify groups are different
|
||||
assert group_id_1 != group_id_2
|
||||
assert res1["data"]["name"] == group_name_1
|
||||
assert res2["data"]["name"] == group_name_2
|
||||
assert res1["data"]["tenant_id"] == tenant_id
|
||||
assert res2["data"]["tenant_id"] == tenant_id
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_create_group_duplicate_name_same_tenant(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with duplicate name in the same tenant."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create first group
|
||||
group_name: str = f"Duplicate Group {uuid.uuid4().hex[:8]}"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res1: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res1["code"] == 0, res1
|
||||
|
||||
# Try to create another group with the same name in the same tenant
|
||||
res2: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res2["code"] == 102 # DATA_ERROR
|
||||
assert "already exists" in res2["message"].lower()
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_create_group_with_whitespace_name(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with whitespace-only name."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Try to create group with whitespace-only name
|
||||
group_payload: dict[str, str] = {
|
||||
"name": " ",
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
# Should fail validation
|
||||
assert res["code"] == 101
|
||||
assert "name" in res["message"].lower() or "empty" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_create_group_special_characters_in_name(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with special characters in name."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create group with special characters
|
||||
group_name: str = f"Group-{uuid.uuid4().hex[:8]}_Test!"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
# Should succeed if special chars are allowed
|
||||
assert res["code"] in (0, 101)
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_create_group_empty_payload(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with empty payload."""
|
||||
group_payload: dict[str, Any] = {}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
assert res["code"] == 101
|
||||
assert "required" in res["message"].lower() or "name" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p3
|
||||
def test_create_group_unicode_name(
|
||||
self, web_api_auth: RAGFlowWebApiAuth
|
||||
) -> None:
|
||||
"""Test creating a group with unicode characters in name."""
|
||||
# First create a team
|
||||
team_name: str = f"Test Team {uuid.uuid4().hex[:8]}"
|
||||
team_payload: dict[str, str] = {"name": team_name}
|
||||
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
|
||||
assert team_res["code"] == 0, team_res
|
||||
tenant_id: str = team_res["data"]["id"]
|
||||
|
||||
# Create group with unicode name
|
||||
group_name: str = f"组{uuid.uuid4().hex[:8]}"
|
||||
group_payload: dict[str, str] = {
|
||||
"name": group_name,
|
||||
"tenant_id": tenant_id,
|
||||
}
|
||||
res: dict[str, Any] = create_group(web_api_auth, group_payload)
|
||||
# Should succeed if unicode is supported
|
||||
assert res["code"] in (0, 101)
|
||||
|
||||
Loading…
Add table
Reference in a new issue