gmakstutis/ragflow
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[OND211-2329]: Updated tets, fixed typings.

Browse source
This commit is contained in:
Hetavi Shah 2025-11-19 16:33:31 +05:30
parent 1ce52f07ff
commit 5af04fdf45
3 changed files with 114 additions and 114 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
test/testcases/test_http_api/test_team_management/test_create_team.py
View file

@ -47,7 +47,7 @@ class TestAuthorization:
invalid_auth: RAGFlowWebApiAuth | None,
expected_code: int,
expected_message: str,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
) -> None:
"""Test team creation with invalid or missing authentication."""
# Try to create team with invalid auth
@ -66,7 +66,7 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_team_with_name_and_user_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with name and user_id."""
# Create team (user_id is optional, defaults to current authenticated user)
@ -74,7 +74,7 @@ class TestTeamCreate:
team_payload: dict[str, str] = {
"name": team_name,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 0, res
assert "data" in res
assert res["data"]["name"] == team_name
@ -85,12 +85,12 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_team_missing_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team without name."""
# Try to create team without name
team_payload: dict[str, str] = {}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 101
assert "name" in res["message"].lower() or "required" in res[
"message"
@ -98,12 +98,12 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_team_empty_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with empty name."""
# Try to create team with empty name
team_payload: dict[str, str] = {"name": ""}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 101
assert "name" in res["message"].lower() or "required" in res[
"message"
@ -111,36 +111,36 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_team_name_too_long(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with name exceeding 100 characters."""
# Try to create team with name too long
long_name: str = "A" * 101
team_payload: dict[str, str] = {"name": long_name}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 101
assert "100" in res["message"] or "length" in res["message"].lower()
@pytest.mark.p1
def test_create_team_invalid_user_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with non-existent user_id."""
team_payload: dict[str, str] = {
"name": "Test Team Invalid User",
"user_id": "non_existent_user_id_12345",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 102
assert "not found" in res["message"].lower()
@pytest.mark.p1
def test_create_team_missing_user_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team without user_id (should use current authenticated user)."""
team_payload: dict[str, str] = {"name": "Test Team No User"}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed since user_id defaults to current authenticated user
assert res["code"] == 0
assert "data" in res
@ -149,7 +149,7 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_team_response_structure(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that team creation returns the expected response structure."""
# Create team
@ -157,7 +157,7 @@ class TestTeamCreate:
team_payload: dict[str, str] = {
"name": team_name,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 0
assert "data" in res
assert isinstance(res["data"], dict)
@ -170,7 +170,7 @@ class TestTeamCreate:
@pytest.mark.p1
def test_create_multiple_teams_same_user(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating multiple teams for the same user."""
# Create first team
@ -178,7 +178,7 @@ class TestTeamCreate:
team_payload_1: dict[str, str] = {
"name": team_name_1,
}
res1: dict[str, Any] = create_team(WebApiAuth, team_payload_1)
res1: dict[str, Any] = create_team(web_api_auth, team_payload_1)
assert res1["code"] == 0, res1
team_id_1: str = res1["data"]["id"]
@ -187,7 +187,7 @@ class TestTeamCreate:
team_payload_2: dict[str, str] = {
"name": team_name_2,
}
res2: dict[str, Any] = create_team(WebApiAuth, team_payload_2)
res2: dict[str, Any] = create_team(web_api_auth, team_payload_2)
assert res2["code"] == 0, res2
team_id_2: str = res2["data"]["id"]
@ -198,14 +198,14 @@ class TestTeamCreate:
@pytest.mark.p2
def test_create_team_with_whitespace_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with whitespace-only name."""
# Try to create team with whitespace-only name
team_payload: dict[str, str] = {
"name": " ",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should fail validation
assert res["code"] == 101
assert "name" in res["message"].lower() or "required" in res[
@ -214,7 +214,7 @@ class TestTeamCreate:
@pytest.mark.p2
def test_create_team_special_characters_in_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with special characters in name."""
# Create team with special characters
@ -222,17 +222,17 @@ class TestTeamCreate:
team_payload: dict[str, str] = {
"name": team_name,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed if special chars are allowed
assert res["code"] in (0, 101)
@pytest.mark.p2
def test_create_team_empty_payload(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with empty payload."""
team_payload: dict[str, Any] = {}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 101
assert "required" in res["message"].lower() or "name" in res[
"message"
@ -240,7 +240,7 @@ class TestTeamCreate:
@pytest.mark.p3
def test_create_team_unicode_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a team with unicode characters in name."""
# Create team with unicode name
@ -248,7 +248,7 @@ class TestTeamCreate:
team_payload: dict[str, str] = {
"name": team_name,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed if unicode is supported
assert res["code"] in (0, 101)

70
test/testcases/test_http_api/test_team_management/test_team_advanced.py
View file

@ -31,7 +31,7 @@ class TestTeamAdvanced:
"""Advanced team management tests."""
def test_team_creation_with_custom_models(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating team with custom model configurations."""
team_name: str = f"Custom Models Team {uuid.uuid4().hex[:8]}"
@ -44,7 +44,7 @@ class TestTeamAdvanced:
# "llm_id": "custom_llm_id",
# "embd_id": "custom_embd_id",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed with defaults if custom models not specified
assert res["code"] == 0, res
@ -52,12 +52,12 @@ class TestTeamAdvanced:
assert "id" in res["data"]
def test_team_creation_response_structure(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that team creation returns complete response structure."""
team_name: str = f"Structure Test Team {uuid.uuid4().hex[:8]}"
team_payload: dict[str, str] = {"name": team_name}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 0, res
assert "data" in res
@ -74,18 +74,18 @@ class TestTeamAdvanced:
assert len(res["data"]["owner_id"]) > 0, "Owner ID should not be empty"
def test_multiple_teams_same_name_allowed(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that multiple teams can have the same name."""
team_name: str = f"Duplicate Name {uuid.uuid4().hex[:8]}"
# Create first team
res1: dict[str, Any] = create_team(WebApiAuth, {"name": team_name})
res1: dict[str, Any] = create_team(web_api_auth, {"name": team_name})
assert res1["code"] == 0, res1
team_id_1: str = res1["data"]["id"]
# Create second team with same name
res2: dict[str, Any] = create_team(WebApiAuth, {"name": team_name})
res2: dict[str, Any] = create_team(web_api_auth, {"name": team_name})
assert res2["code"] == 0, res2
team_id_2: str = res2["data"]["id"]
@ -96,7 +96,7 @@ class TestTeamAdvanced:
)
def test_team_creation_with_credit_limit(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating team with custom credit limit."""
team_name: str = f"Credit Test Team {uuid.uuid4().hex[:8]}"
@ -106,14 +106,14 @@ class TestTeamAdvanced:
"name": team_name,
"credit": custom_credit,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed
assert res["code"] == 0, res
# Note: Credit may not be in response, but should be set internally
def test_team_name_with_special_characters(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team names with special characters."""
special_names: list[str] = [
@ -124,7 +124,7 @@ class TestTeamAdvanced:
]
for name in special_names:
res: dict[str, Any] = create_team(WebApiAuth, {"name": name})
res: dict[str, Any] = create_team(web_api_auth, {"name": name})
# Should either accept or reject with clear message
if res["code"] == 0:
assert res["data"]["name"] == name, (
@ -134,26 +134,26 @@ class TestTeamAdvanced:
# (Current implementation accepts special chars)
def test_team_creation_default_owner(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that team creator is set as owner by default."""
team_name: str = f"Owner Test Team {uuid.uuid4().hex[:8]}"
res: dict[str, Any] = create_team(WebApiAuth, {"name": team_name})
res: dict[str, Any] = create_team(web_api_auth, {"name": team_name})
assert res["code"] == 0, res
assert "owner_id" in res["data"], "Owner ID should be in response"
# Owner should be the authenticated user
# (Cannot verify without knowing WebApiAuth user ID)
# (Cannot verify without knowing web_api_auth user ID)
def test_concurrent_team_creation(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test concurrent team creation."""
import concurrent.futures
def create_test_team(index: int) -> dict[str, Any]:
team_name: str = f"Concurrent Team {index}_{uuid.uuid4().hex[:8]}"
return create_team(WebApiAuth, {"name": team_name})
return create_team(web_api_auth, {"name": team_name})
# Create 10 teams concurrently
count: int = 10
@ -178,7 +178,7 @@ class TestTeamAdvanced:
)
def test_team_with_invalid_model_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with invalid model ID."""
team_name: str = f"Invalid Model Team {uuid.uuid4().hex[:8]}"
@ -186,7 +186,7 @@ class TestTeamAdvanced:
"name": team_name,
"llm_id": "invalid_nonexistent_model_id_12345",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should reject with clear error message
assert res["code"] != 0, "Invalid model ID should be rejected"
@ -197,7 +197,7 @@ class TestTeamAdvanced:
), "Error message should mention model issue"
def test_team_creation_with_negative_credit(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with negative credit."""
team_name: str = f"Negative Credit Team {uuid.uuid4().hex[:8]}"
@ -205,7 +205,7 @@ class TestTeamAdvanced:
"name": team_name,
"credit": -100,
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should reject negative credit
assert res["code"] != 0, "Negative credit should be rejected"
@ -214,10 +214,10 @@ class TestTeamAdvanced:
)
def test_team_creation_empty_json_payload(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with completely empty payload."""
res: dict[str, Any] = create_team(WebApiAuth, {})
res: dict[str, Any] = create_team(web_api_auth, {})
# Should reject with clear error
assert res["code"] != 0, "Empty payload should be rejected"
@ -227,7 +227,7 @@ class TestTeamAdvanced:
), "Error should mention missing name"
def test_team_unicode_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with full unicode name."""
unicode_names: list[str] = [
@ -239,7 +239,7 @@ class TestTeamAdvanced:
]
for name in unicode_names:
res: dict[str, Any] = create_team(WebApiAuth, {"name": name})
res: dict[str, Any] = create_team(web_api_auth, {"name": name})
# Should handle unicode properly
if res["code"] == 0:
@ -250,7 +250,7 @@ class TestTeamAdvanced:
@pytest.mark.p3
def test_team_creation_with_all_optional_params(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with all optional parameters."""
team_name: str = f"Full Params Team {uuid.uuid4().hex[:8]}"
@ -265,7 +265,7 @@ class TestTeamAdvanced:
# "img2txt_id": "valid_img2txt_id",
# "rerank_id": "valid_rerank_id",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should succeed
assert res["code"] == 0, res
@ -273,12 +273,12 @@ class TestTeamAdvanced:
@pytest.mark.p3
def test_team_max_name_length(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team with maximum allowed name length."""
# API spec says max 100 characters
max_name: str = "A" * 100
res: dict[str, Any] = create_team(WebApiAuth, {"name": max_name})
res: dict[str, Any] = create_team(web_api_auth, {"name": max_name})
# Should accept 100 characters
assert res["code"] == 0, "100-character name should be accepted"
@ -286,12 +286,12 @@ class TestTeamAdvanced:
@pytest.mark.p3
def test_team_name_just_over_limit(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team with name just over limit."""
# 101 characters (1 over limit)
long_name: str = "A" * 101
res: dict[str, Any] = create_team(WebApiAuth, {"name": long_name})
res: dict[str, Any] = create_team(web_api_auth, {"name": long_name})
# Should reject
assert res["code"] != 0, "101-character name should be rejected"
@ -303,15 +303,15 @@ class TestTeamAdvanced:
@pytest.mark.p3
def test_team_creation_idempotency(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that repeated team creation creates separate teams."""
team_name: str = f"Idempotency Test {uuid.uuid4().hex[:8]}"
payload: dict[str, str] = {"name": team_name}
# Create same team twice
res1: dict[str, Any] = create_team(WebApiAuth, payload)
res2: dict[str, Any] = create_team(WebApiAuth, payload)
res1: dict[str, Any] = create_team(web_api_auth, payload)
res2: dict[str, Any] = create_team(web_api_auth, payload)
# Both should succeed and create different teams
assert res1["code"] == 0, res1
@ -322,7 +322,7 @@ class TestTeamAdvanced:
@pytest.mark.p3
def test_team_with_parser_ids(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test team creation with custom parser IDs."""
team_name: str = f"Parser Test {uuid.uuid4().hex[:8]}"
@ -331,7 +331,7 @@ class TestTeamAdvanced:
"name": team_name,
"parser_ids": "naive,qa,table,paper,book,laws,presentation,manual,wiki",
}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
# Should accept valid parser IDs
assert res["code"] == 0, res

106
test/testcases/test_http_api/test_team_management/test_team_users.py
View file

@ -50,12 +50,12 @@ class TestAddUsersAuthorization:
self,
invalid_auth: RAGFlowWebApiAuth | None,
expected_code: int,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
) -> None:
"""Test adding users with invalid or missing authentication."""
# Create a team first
team_payload: dict[str, str] = {"name": f"Test Team {uuid.uuid4().hex[:8]}"}
team_res: dict[str, Any] = create_team(WebApiAuth, team_payload)
team_res: dict[str, Any] = create_team(web_api_auth, team_payload)
tenant_id: str = team_res["data"]["id"]
# Try to add users with invalid auth
@ -69,15 +69,15 @@ class TestAddUsers:
"""Comprehensive tests for adding users to a team."""
@pytest.fixture
def test_team(self, WebApiAuth: RAGFlowWebApiAuth) -> dict[str, Any]:
def test_team(self, web_api_auth: RAGFlowWebApiAuth) -> dict[str, Any]:
"""Create a test team for use in tests."""
team_payload: dict[str, str] = {"name": f"Test Team {uuid.uuid4().hex[:8]}"}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_users(self, WebApiAuth: RAGFlowWebApiAuth) -> list[dict[str, Any]]:
def test_users(self, web_api_auth: RAGFlowWebApiAuth) -> list[dict[str, Any]]:
"""Create test users for use in tests."""
users = []
for i in range(5):
@ -88,21 +88,21 @@ class TestAddUsers:
"password": password,
"nickname": f"Test User {i}",
}
user_res: dict[str, Any] = create_user(WebApiAuth, user_payload)
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"]})
return users
@pytest.mark.p1
def test_add_single_user_with_email_string(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding a single user using email string format."""
tenant_id: str = test_team["id"]
user_email: str = test_users[0]["email"]
add_payload: dict[str, list[str]] = {"users": [user_email]}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0
assert "data" in res
@ -115,7 +115,7 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_single_user_with_role(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding a single user with admin role."""
tenant_id: str = test_team["id"]
@ -124,7 +124,7 @@ class TestAddUsers:
add_payload: dict[str, list[dict[str, str]]] = {
"users": [{"email": user_email, "role": "admin"}]
}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0
assert len(res["data"]["added"]) == 1
@ -133,14 +133,14 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_multiple_users(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding multiple users in bulk."""
tenant_id: str = test_team["id"]
user_emails: list[str] = [user["email"] for user in test_users[:3]]
add_payload: dict[str, list[str]] = {"users": user_emails}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0
assert len(res["data"]["added"]) == 3
@ -150,7 +150,7 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_users_mixed_formats(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding users with mixed string and object formats."""
tenant_id: str = test_team["id"]
@ -162,7 +162,7 @@ class TestAddUsers:
test_users[2]["email"], # String format
]
}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0
assert len(res["data"]["added"]) == 3
@ -172,14 +172,14 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_user_unregistered_email(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test adding a user with unregistered email."""
tenant_id: str = test_team["id"]
unregistered_email: str = f"unregistered_{uuid.uuid4().hex[:8]}@example.com"
add_payload: dict[str, list[str]] = {"users": [unregistered_email]}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 102 # DATA_ERROR
assert len(res["data"]["added"]) == 0
@ -188,7 +188,7 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_user_already_member(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding a user who is already a member of the team."""
tenant_id: str = test_team["id"]
@ -196,11 +196,11 @@ class TestAddUsers:
# Add user first time
add_payload: dict[str, list[str]] = {"users": [user_email]}
res1: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res1: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res1["code"] == 0
# Try to add same user again
res2: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res2: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res2["code"] == 0 # Returns success but with failed entry
assert len(res2["data"]["added"]) == 0
assert len(res2["data"]["failed"]) == 1
@ -208,7 +208,7 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_users_partial_success(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding users where some succeed and some fail."""
tenant_id: str = test_team["id"]
@ -217,7 +217,7 @@ class TestAddUsers:
add_payload: dict[str, list[str]] = {
"users": [test_users[0]["email"], unregistered_email, test_users[1]["email"]]
}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0
assert len(res["data"]["added"]) == 2
@ -226,30 +226,30 @@ class TestAddUsers:
@pytest.mark.p1
def test_add_users_empty_list(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test adding users with empty list."""
tenant_id: str = test_team["id"]
add_payload: dict[str, list[str]] = {"users": []}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 101 # ARGUMENT_ERROR
assert "non-empty" in res["message"].lower() or "empty" in res["message"].lower()
@pytest.mark.p1
def test_add_users_missing_users_field(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test adding users without 'users' field."""
tenant_id: str = test_team["id"]
add_payload: dict[str, Any] = {}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 101 # ARGUMENT_ERROR
@pytest.mark.p1
def test_add_users_invalid_role(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test adding user with invalid role."""
tenant_id: str = test_team["id"]
@ -258,7 +258,7 @@ class TestAddUsers:
add_payload: dict[str, list[dict[str, str]]] = {
"users": [{"email": user_email, "role": "invalid_role"}]
}
res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert res["code"] == 0 # Returns success but with failed entry
assert len(res["data"]["added"]) == 0
@ -267,7 +267,7 @@ class TestAddUsers:
@pytest.mark.p2
def test_add_users_not_owner_or_admin(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test that non-admin/non-owner users cannot add users."""
tenant_id: str = test_team["id"]
@ -275,7 +275,7 @@ class TestAddUsers:
# Add a normal user to the team
add_payload: dict[str, list[str]] = {"users": [user_email]}
add_res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
add_res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert add_res["code"] == 0
# Create auth for the normal user (this would require getting their token)
@ -289,15 +289,15 @@ class TestRemoveUsers:
"""Comprehensive tests for removing users from a team."""
@pytest.fixture
def test_team(self, WebApiAuth: RAGFlowWebApiAuth) -> dict[str, Any]:
def test_team(self, web_api_auth: RAGFlowWebApiAuth) -> dict[str, Any]:
"""Create a test team for use in tests."""
team_payload: dict[str, str] = {"name": f"Test Team {uuid.uuid4().hex[:8]}"}
res: dict[str, Any] = create_team(WebApiAuth, team_payload)
res: dict[str, Any] = create_team(web_api_auth, team_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_users(self, WebApiAuth: RAGFlowWebApiAuth) -> list[dict[str, Any]]:
def test_users(self, web_api_auth: RAGFlowWebApiAuth) -> list[dict[str, Any]]:
"""Create test users for use in tests."""
users = []
for i in range(5):
@ -308,21 +308,21 @@ class TestRemoveUsers:
"password": password,
"nickname": f"Test User {i}",
}
user_res: dict[str, Any] = create_user(WebApiAuth, user_payload)
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"]})
return users
@pytest.fixture
def team_with_users(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> dict[str, Any]:
"""Create a team with users already added."""
tenant_id: str = test_team["id"]
user_emails: list[str] = [user["email"] for user in test_users[:3]]
add_payload: dict[str, list[str]] = {"users": user_emails}
add_res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
add_res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert add_res["code"] == 0
return {
@ -332,14 +332,14 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_single_user(
self, WebApiAuth: RAGFlowWebApiAuth, team_with_users: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, team_with_users: dict[str, Any]
) -> None:
"""Test removing a single user from a team."""
tenant_id: str = team_with_users["team"]["id"]
user_id: str = team_with_users["users"][0]["id"]
remove_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0
assert "data" in res
@ -351,14 +351,14 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_multiple_users(
self, WebApiAuth: RAGFlowWebApiAuth, team_with_users: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, team_with_users: dict[str, Any]
) -> None:
"""Test removing multiple users in bulk."""
tenant_id: str = team_with_users["team"]["id"]
user_ids: list[str] = [user["id"] for user in team_with_users["users"][:2]]
remove_payload: dict[str, list[str]] = {"user_ids": user_ids}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0
assert len(res["data"]["removed"]) == 2
@ -368,7 +368,7 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_user_not_in_team(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test removing a user who is not a member of the team."""
tenant_id: str = test_team["id"]
@ -376,7 +376,7 @@ class TestRemoveUsers:
user_id: str = test_users[3]["id"]
remove_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0 # Returns success but with failed entry
assert len(res["data"]["removed"]) == 0
@ -385,14 +385,14 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_owner(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test that owner cannot be removed."""
tenant_id: str = test_team["id"]
owner_id: str = test_team["owner_id"]
remove_payload: dict[str, list[str]] = {"user_ids": [owner_id]}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0 # Returns success but with failed entry
assert len(res["data"]["removed"]) == 0
@ -401,7 +401,7 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_users_partial_success(
self, WebApiAuth: RAGFlowWebApiAuth, team_with_users: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, team_with_users: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test removing users where some succeed and some fail."""
tenant_id: str = team_with_users["team"]["id"]
@ -410,7 +410,7 @@ class TestRemoveUsers:
invalid_user_id: str = test_users[3]["id"] # Not in team
remove_payload: dict[str, list[str]] = {"user_ids": [valid_user_id, invalid_user_id]}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0
assert len(res["data"]["removed"]) == 1
@ -420,36 +420,36 @@ class TestRemoveUsers:
@pytest.mark.p1
def test_remove_users_empty_list(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test removing users with empty list."""
tenant_id: str = test_team["id"]
remove_payload: dict[str, list[str]] = {"user_ids": []}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 101 # ARGUMENT_ERROR
assert "non-empty" in res["message"].lower() or "empty" in res["message"].lower()
@pytest.mark.p1
def test_remove_users_missing_user_ids_field(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test removing users without 'user_ids' field."""
tenant_id: str = test_team["id"]
remove_payload: dict[str, Any] = {}
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 101 # ARGUMENT_ERROR
@pytest.mark.p1
def test_remove_users_invalid_user_id_format(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> None:
"""Test removing users with invalid user ID format."""
tenant_id: str = test_team["id"]
remove_payload: dict[str, list[Any]] = {"user_ids": [12345]} # Not a string
res: dict[str, Any] = remove_users_from_team(WebApiAuth, tenant_id, remove_payload)
res: dict[str, Any] = remove_users_from_team(web_api_auth, tenant_id, remove_payload)
assert res["code"] == 0 # Returns success but with failed entry
assert len(res["data"]["removed"]) == 0
assert len(res["data"]["failed"]) == 1
@ -457,7 +457,7 @@ class TestRemoveUsers:
@pytest.mark.p2
def test_remove_last_admin(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any], test_users: list[dict[str, Any]]
) -> None:
"""Test that the last admin cannot remove themselves."""
tenant_id: str = test_team["id"]
@ -467,7 +467,7 @@ class TestRemoveUsers:
add_payload: dict[str, list[dict[str, str]]] = {
"users": [{"email": user_email, "role": "admin"}]
}
add_res: dict[str, Any] = add_users_to_team(WebApiAuth, tenant_id, add_payload)
add_res: dict[str, Any] = add_users_to_team(web_api_auth, tenant_id, add_payload)
assert add_res["code"] == 0
admin_user_id: str = add_res["data"]["added"][0]["id"]