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

[OND211-2329]: Added few tests for department API and updated typings.

Browse source
This commit is contained in:
Hetavi Shah 2025-11-18 18:47:36 +05:30
parent 26a17e6250
commit c38ec24b08
7 changed files with 1627 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
api/apps/department_app.py
View file

@ -22,7 +22,7 @@ from flask import Blueprint, Response, request
from flask_login import current_user, login_required
from api.db import UserTenantRole
from api.db.db_models import Department, Tenant, User, UserDepartment, UserTenant
from api.db.db_models import DB, Department, Tenant, User, UserDepartment, UserTenant
from api.db.services.user_service import (
DepartmentService,
TenantService,
@ -324,8 +324,6 @@ def update_department(department_id: str) -> Response:
)
try:
from api.db.db_models import DB
# Update the department
with DB.connection_context():
Department.update(update_data).where(
@ -420,8 +418,6 @@ def delete_department(department_id: str) -> Response:
)
try:
from api.db.db_models import DB
# Soft delete the department and all related user_department records
with DB.connection_context():
# Soft delete all user-department relationships for this department
@ -672,8 +668,6 @@ def remove_member(department_id: str, user_id: str) -> Response:
)
try:
from api.db.db_models import DB
# Check if user is in the department
user_department: Optional[UserDepartment] = UserDepartmentService.filter_by_department_and_user_id(
department_id, user_id

138
test/testcases/test_http_api/test_department_management/test_add_members.py
View file

@ -26,6 +26,7 @@ from common import (
create_department,
create_team,
create_user,
login_as_user,
)
from configs import INVALID_API_TOKEN
from libs.auth import RAGFlowWebApiAuth
@ -52,13 +53,13 @@ class TestAuthorization:
invalid_auth: RAGFlowWebApiAuth | None,
expected_code: int,
expected_message: str,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
) -> None:
"""Test adding members with invalid or missing authentication."""
# Create a team and department first
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(WebApiAuth, team_payload)
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")
@ -69,7 +70,7 @@ class TestAuthorization:
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] != 0:
pytest.skip("Department creation failed, skipping auth test")
@ -88,28 +89,28 @@ class TestAddMembers:
"""Comprehensive tests for adding members to a department."""
@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_department(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> dict[str, Any]:
"""Create a test department for use in tests."""
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": test_team["id"],
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_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(3):
@ -119,7 +120,7 @@ class TestAddMembers:
"password": "TestPassword123!",
"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
@ -127,20 +128,20 @@ class TestAddMembers:
@pytest.fixture
def team_with_users(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_team: dict[str, Any],
test_users: list[dict[str, Any]],
) -> dict[str, Any]:
"""Add test users to the team."""
for user in test_users:
add_payload: dict[str, list[str]] = {"users": [user["email"]]}
add_users_to_team(WebApiAuth, test_team["id"], add_payload)
add_users_to_team(web_api_auth, test_team["id"], add_payload)
return test_team
@pytest.mark.p1
def test_add_single_member(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
@ -152,7 +153,7 @@ class TestAddMembers:
user_id: str = test_users[0]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0, res
@ -166,7 +167,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_multiple_members(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
@ -178,7 +179,7 @@ class TestAddMembers:
user_ids: list[str] = [user["id"] for user in test_users[:2]]
add_payload: dict[str, list[str]] = {"user_ids": user_ids}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0, res
@ -189,13 +190,13 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_missing_user_ids(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test adding members without user_ids."""
add_payload: dict[str, Any] = {}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 101
@ -206,13 +207,13 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_empty_user_ids(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test adding members with empty user_ids array."""
add_payload: dict[str, list[str]] = {"user_ids": []}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 101
@ -223,7 +224,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_invalid_user_id(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test adding a non-existent user."""
@ -231,7 +232,7 @@ class TestAddMembers:
"user_ids": ["non_existent_user_id_12345"]
}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0 # API returns success with failed list
@ -243,7 +244,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_user_not_in_team(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
@ -258,14 +259,14 @@ class TestAddMembers:
"password": "TestPassword123!",
"nickname": "Not In Team User",
}
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:
pytest.skip("User creation failed")
user_id: str = user_res["data"]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0 # API returns success with failed list
@ -276,7 +277,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_duplicate_member(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
@ -290,14 +291,14 @@ class TestAddMembers:
# Add user first time
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res1: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res1["code"] == 0
assert len(res1["data"]["added"]) == 1
# Try to add same user again
res2: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res2["code"] == 0 # API returns success with failed list
assert len(res2["data"]["added"]) == 0
@ -307,7 +308,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_invalid_department_id(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
@ -318,7 +319,7 @@ class TestAddMembers:
user_id: str = test_users[0]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = add_department_members(
WebApiAuth, "non_existent_department_id_12345", add_payload
web_api_auth, "non_existent_department_id_12345", add_payload
)
assert res["code"] == 102
@ -329,13 +330,13 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_invalid_user_id_format(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test adding members with invalid user ID formats."""
add_payload: dict[str, list[Any]] = {"user_ids": ["", " ", 123, None]}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0 # API returns success with failed list
@ -350,7 +351,7 @@ class TestAddMembers:
@pytest.mark.p1
def test_add_member_mixed_valid_invalid(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
@ -365,7 +366,7 @@ class TestAddMembers:
"user_ids": [valid_user_id, invalid_user_id]
}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0
@ -376,18 +377,75 @@ class TestAddMembers:
@pytest.mark.p2
def test_add_member_not_team_owner_or_admin(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test adding members when user is not team owner or admin."""
# This test would require creating a team with a different user
# and then trying to add members as a non-admin user
# For now, we'll skip this as it requires multi-user setup
pytest.skip("Requires multi-user setup to test permission restrictions")
# Create a team (current user is owner)
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, "Failed to create team"
tenant_id: str = team_res["data"]["id"]
# Create a department
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, "Failed to create department"
department_id: str = dept_res["data"]["id"]
# Create a normal user (not admin/owner)
normal_user_email: str = f"normaluser_{uuid.uuid4().hex[:8]}@example.com"
normal_user_password: str = "TestPassword123!"
normal_user_payload: dict[str, str] = {
"email": normal_user_email,
"password": normal_user_password,
"nickname": "Normal User",
}
normal_user_res: dict[str, Any] = create_user(web_api_auth, normal_user_payload)
assert normal_user_res["code"] == 0, "Failed to create normal user"
# Add the normal user to the team as a normal member (not admin)
add_team_payload: dict[str, list[str]] = {"users": [normal_user_email]}
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"
# 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!"
another_user_payload: dict[str, str] = {
"email": another_user_email,
"password": another_user_password,
"nickname": "Another User",
}
another_user_res: dict[str, Any] = create_user(web_api_auth, another_user_payload)
assert another_user_res["code"] == 0, "Failed to create another user"
another_user_id: str = another_user_res["data"]["id"]
# Add another user to the team
add_another_team_payload: dict[str, list[str]] = {"users": [another_user_email]}
add_another_team_res: dict[str, Any] = add_users_to_team(
web_api_auth, tenant_id, add_another_team_payload
)
assert add_another_team_res["code"] == 0, "Failed to add another user to team"
# Login as the normal user (not admin/owner)
normal_user_auth: RAGFlowWebApiAuth = login_as_user(normal_user_email, normal_user_password)
# Try to add a member as the normal user (should fail)
add_payload: dict[str, list[str]] = {"user_ids": [another_user_id]}
add_res: dict[str, Any] = add_department_members(
normal_user_auth, department_id, add_payload
)
assert add_res["code"] == 108, f"Expected permission error (108), got {add_res}"
assert "only team owners or admins" in add_res["message"].lower()
@pytest.mark.p2
def test_add_member_response_structure(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
@ -399,7 +457,7 @@ class TestAddMembers:
user_id: str = test_users[0]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert res["code"] == 0

80
test/testcases/test_http_api/test_department_management/test_create_department.py
View file

@ -47,13 +47,13 @@ class TestAuthorization:
invalid_auth: RAGFlowWebApiAuth | None,
expected_code: int,
expected_message: str,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
) -> None:
"""Test department 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(WebApiAuth, team_payload)
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")
@ -76,13 +76,13 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_with_name_and_tenant_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -92,7 +92,7 @@ class TestDepartmentCreate:
"name": dept_name,
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0, res
assert "data" in res
assert res["data"]["name"] == dept_name
@ -102,13 +102,13 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_with_description(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -120,26 +120,26 @@ class TestDepartmentCreate:
"tenant_id": tenant_id,
"description": description,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0, res
assert res["data"]["name"] == dept_name
assert res["data"]["description"] == description
@pytest.mark.p1
def test_create_department_missing_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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 department without name
dept_payload: dict[str, str] = {"tenant_id": tenant_id}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 101
assert "name" in res["message"].lower() or "required" in res[
"message"
@ -147,19 +147,19 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_empty_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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 department with empty name
dept_payload: dict[str, str] = {"name": "", "tenant_id": tenant_id}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 101
assert "name" in res["message"].lower() or "empty" in res[
"message"
@ -167,12 +167,12 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_missing_tenant_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department without tenant_id."""
# Try to create department without tenant_id
dept_payload: dict[str, str] = {"name": "Test Department"}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 101
assert "tenant_id" in res["message"].lower() or "required" in res[
"message"
@ -180,14 +180,14 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_invalid_tenant_id(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department with non-existent tenant_id."""
dept_payload: dict[str, str] = {
"name": "Test Department Invalid Tenant",
"tenant_id": "non_existent_tenant_id_12345",
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_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
@ -197,13 +197,13 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_name_too_long(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -215,7 +215,7 @@ class TestDepartmentCreate:
"name": long_name,
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
# API doesn't validate length, so it may succeed or fail at database level
# If it succeeds, the name might be truncated; if it fails, we get an exception error
assert res["code"] in (0, 100) # Success or exception error
@ -226,7 +226,7 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_not_team_owner_or_admin(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department when user is not team owner or admin."""
# This test would require creating a team with a different user
@ -236,13 +236,13 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_department_response_structure(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that department 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(WebApiAuth, team_payload)
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"]
@ -252,7 +252,7 @@ class TestDepartmentCreate:
"name": dept_name,
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0
assert "data" in res
assert isinstance(res["data"], dict)
@ -266,13 +266,13 @@ class TestDepartmentCreate:
@pytest.mark.p1
def test_create_multiple_departments_same_team(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating multiple departments 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(WebApiAuth, team_payload)
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"]
@ -282,7 +282,7 @@ class TestDepartmentCreate:
"name": dept_name_1,
"tenant_id": tenant_id,
}
res1: dict[str, Any] = create_department(WebApiAuth, dept_payload_1)
res1: dict[str, Any] = create_department(web_api_auth, dept_payload_1)
assert res1["code"] == 0, res1
dept_id_1: str = res1["data"]["id"]
@ -292,7 +292,7 @@ class TestDepartmentCreate:
"name": dept_name_2,
"tenant_id": tenant_id,
}
res2: dict[str, Any] = create_department(WebApiAuth, dept_payload_2)
res2: dict[str, Any] = create_department(web_api_auth, dept_payload_2)
assert res2["code"] == 0, res2
dept_id_2: str = res2["data"]["id"]
@ -305,13 +305,13 @@ class TestDepartmentCreate:
@pytest.mark.p2
def test_create_department_with_whitespace_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -320,7 +320,7 @@ class TestDepartmentCreate:
"name": " ",
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
# Should fail validation
assert res["code"] == 101
assert "name" in res["message"].lower() or "empty" in res[
@ -329,13 +329,13 @@ class TestDepartmentCreate:
@pytest.mark.p2
def test_create_department_special_characters_in_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -345,17 +345,17 @@ class TestDepartmentCreate:
"name": dept_name,
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
# Should succeed if special chars are allowed
assert res["code"] in (0, 101)
@pytest.mark.p2
def test_create_department_empty_payload(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department with empty payload."""
dept_payload: dict[str, Any] = {}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 101
assert "required" in res["message"].lower() or "name" in res[
"message"
@ -363,13 +363,13 @@ class TestDepartmentCreate:
@pytest.mark.p3
def test_create_department_unicode_name(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test creating a department 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(WebApiAuth, team_payload)
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"]
@ -379,7 +379,7 @@ class TestDepartmentCreate:
"name": dept_name,
"tenant_id": tenant_id,
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
# Should succeed if unicode is supported
assert res["code"] in (0, 101)

364
test/testcases/test_http_api/test_department_management/test_delete_department.py Normal file
View file

@ -0,0 +1,364 @@
#
# 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 uuid
from typing import Any
import pytest
from common import (
add_department_members,
add_users_to_team,
create_department,
create_team,
create_user,
delete_department,
get_user_info,
login_as_user,
)
from configs import INVALID_API_TOKEN
from libs.auth import RAGFlowWebApiAuth
# ---------------------------------------------------------------------------
# Test Classes
# ---------------------------------------------------------------------------
@pytest.mark.p1
class TestAuthorization:
"""Tests for authentication behavior when deleting a department."""
@pytest.mark.parametrize(
("invalid_auth", "expected_code", "expected_message"),
[
(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 deleting a department with invalid or missing authentication."""
# Create a team and department first
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"]
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] != 0:
pytest.skip("Department creation failed, skipping auth test")
department_id: str = dept_res["data"]["id"]
# Try to delete department with invalid auth
res: dict[str, Any] = delete_department(invalid_auth, department_id)
assert res["code"] == expected_code, res
if expected_message:
assert expected_message in res["message"]
@pytest.mark.p1
class TestDeleteDepartment:
"""Comprehensive tests for deleting a department."""
@pytest.fixture
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(web_api_auth, team_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_department(
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> dict[str, Any]:
"""Create a test department for use in tests."""
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": test_team["id"],
}
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_department_with_members(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_team: dict[str, Any],
) -> dict[str, Any]:
"""Create a test department and add the current user as a member."""
# Get current user ID
user_info_res: dict[str, Any] = get_user_info(web_api_auth)
if user_info_res["code"] == 0:
user_id: str = user_info_res["data"]["id"]
# Add current user to department (they're already team owner/admin)
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_department_members(web_api_auth, test_department["id"], add_payload)
return test_department
@pytest.mark.p1
def test_delete_department_success(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department_with_members: dict[str, Any],
) -> None:
"""Test successfully deleting a department."""
department_id: str = test_department_with_members["id"]
# Delete the department
res: dict[str, Any] = delete_department(web_api_auth, department_id)
assert res["code"] == 0, res
assert res["data"] is True
assert "deleted successfully" in res["message"].lower()
@pytest.mark.p1
def test_delete_department_invalid_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test deleting a department with an invalid department ID."""
invalid_id: str = f"invalid_{uuid.uuid4().hex[:8]}"
res: dict[str, Any] = delete_department(web_api_auth, invalid_id)
assert res["code"] == 100 # DATA_ERROR
assert "not found" in res["message"].lower()
@pytest.mark.p1
def test_delete_department_not_member(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test deleting a department when user is not a member."""
# Create a new user and add them to the team but not to the department
email: str = f"testuser_{uuid.uuid4().hex[:8]}@example.com"
user_payload: dict[str, str] = {
"email": email,
"password": "TestPassword123!",
"nickname": "Test User",
}
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
if user_res["code"] != 0:
pytest.skip("User creation failed")
user_id: str = user_res["data"]["id"]
# Add user to team
team_id: str = test_department["tenant_id"]
add_payload: dict[str, list[str]] = {"users": [email]}
add_users_to_team(web_api_auth, team_id, add_payload)
# Login as the new user
new_user_auth: RAGFlowWebApiAuth = login_as_user(email, "TestPassword123!")
# Try to delete department (user is not a member)
res: dict[str, Any] = delete_department(new_user_auth, test_department["id"])
assert res["code"] == 103 # PERMISSION_ERROR
assert "member" in res["message"].lower()
@pytest.mark.p1
def test_delete_department_not_team_admin_or_owner(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test deleting a department when user is not team admin or owner."""
# Create a new user
email: str = f"testuser_{uuid.uuid4().hex[:8]}@example.com"
user_payload: dict[str, str] = {
"email": email,
"password": "TestPassword123!",
"nickname": "Test User",
}
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
if user_res["code"] != 0:
pytest.skip("User creation failed")
user_id: str = user_res["data"]["id"]
# Add user to team as normal member
team_id: str = test_department["tenant_id"]
add_payload: dict[str, list[str]] = {"users": [email]}
add_users_to_team(web_api_auth, team_id, add_payload)
# Accept invitation (if needed)
# Note: This depends on the invitation flow implementation
# Add user to department
dept_add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_department_members(web_api_auth, test_department["id"], dept_add_payload)
# Login as the new user (normal member, not admin/owner)
new_user_auth: RAGFlowWebApiAuth = login_as_user(email, "TestPassword123!")
# Try to delete department (user is member but not admin/owner)
res: dict[str, Any] = delete_department(new_user_auth, test_department["id"])
assert res["code"] == 103 # PERMISSION_ERROR
assert "owner" in res["message"].lower() or "admin" in res["message"].lower()
@pytest.mark.p1
def test_delete_department_response_structure(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department_with_members: dict[str, Any],
) -> None:
"""Test that department deletion returns the expected response structure."""
department_id: str = test_department_with_members["id"]
res: dict[str, Any] = delete_department(web_api_auth, department_id)
assert res["code"] == 0
assert "data" in res
assert res["data"] is True
assert "message" in res
assert isinstance(res["message"], str)
assert "deleted successfully" in res["message"].lower()
@pytest.mark.p1
def test_delete_department_already_deleted(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department_with_members: dict[str, Any],
) -> None:
"""Test deleting a department that has already been deleted."""
department_id: str = test_department_with_members["id"]
# Delete the department first
res1: dict[str, Any] = delete_department(web_api_auth, department_id)
assert res1["code"] == 0
# Try to delete again
res2: dict[str, Any] = delete_department(web_api_auth, department_id)
# Should return error (department not found or already deleted)
assert res2["code"] != 0
assert "not found" in res2["message"].lower() or "deleted" in res2["message"].lower()
@pytest.mark.p1
def test_delete_department_with_members(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_team: dict[str, Any],
) -> None:
"""Test deleting a department that has members."""
# Create test users
users = []
for i in range(2):
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 not users:
pytest.skip("User creation failed")
# Add users to team
for user in users:
add_payload: dict[str, list[str]] = {"users": [user["email"]]}
add_users_to_team(web_api_auth, test_team["id"], add_payload)
# Add users to department
user_ids: list[str] = [user["id"] for user in users]
dept_add_payload: dict[str, list[str]] = {"user_ids": user_ids}
add_department_members(web_api_auth, test_department["id"], dept_add_payload)
# Ensure current user is also a member (for deletion permission)
user_info_res: dict[str, Any] = get_user_info(web_api_auth)
if user_info_res["code"] == 0:
current_user_id: str = user_info_res["data"]["id"]
dept_add_payload2: dict[str, list[str]] = {"user_ids": [current_user_id]}
add_department_members(web_api_auth, test_department["id"], dept_add_payload2)
# Delete the department (should also remove all member relationships)
res: dict[str, Any] = delete_department(web_api_auth, test_department["id"])
assert res["code"] == 0, res
assert res["data"] is True
assert "member relationships" in res["message"].lower() or "deleted successfully" in res["message"].lower()
@pytest.mark.p2
def test_delete_multiple_departments(
self,
web_api_auth: RAGFlowWebApiAuth,
test_team: dict[str, Any],
) -> None:
"""Test deleting multiple departments from the same team."""
# Create multiple departments
departments = []
for i in range(3):
dept_name: str = f"Department {i} {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": test_team["id"],
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] == 0:
departments.append(dept_res["data"])
if not departments:
pytest.skip("Department creation failed")
# Add current user to all departments
user_info_res: dict[str, Any] = get_user_info(web_api_auth)
if user_info_res["code"] == 0:
current_user_id: str = user_info_res["data"]["id"]
for dept in departments:
dept_add_payload: dict[str, list[str]] = {"user_ids": [current_user_id]}
add_department_members(web_api_auth, dept["id"], dept_add_payload)
# Delete all departments
for dept in departments:
res: dict[str, Any] = delete_department(web_api_auth, dept["id"])
assert res["code"] == 0, f"Failed to delete department {dept['id']}: {res}"
@pytest.mark.p2
def test_delete_department_empty_string_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test deleting a department with empty string ID."""
res: dict[str, Any] = delete_department(web_api_auth, "")
assert res["code"] != 0
assert "not found" in res["message"].lower() or res["code"] == 100
@pytest.mark.p2
def test_delete_department_special_characters_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test deleting a department with special characters in ID."""
invalid_id: str = "dept-123_!@#$%"
res: dict[str, Any] = delete_department(web_api_auth, invalid_id)
assert res["code"] != 0
assert "not found" in res["message"].lower() or res["code"] == 100

475
test/testcases/test_http_api/test_department_management/test_list_members.py Normal file
View file

@ -0,0 +1,475 @@
#
# 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 uuid
from typing import Any
import pytest
from common import (
add_department_members,
add_users_to_team,
create_department,
create_team,
create_user,
list_department_members,
login_as_user,
)
from configs import INVALID_API_TOKEN
from libs.auth import RAGFlowWebApiAuth
# ---------------------------------------------------------------------------
# Test Classes
# ---------------------------------------------------------------------------
@pytest.mark.p1
class TestAuthorization:
"""Tests for authentication behavior when listing department members."""
@pytest.mark.parametrize(
("invalid_auth", "expected_code", "expected_message"),
[
(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 listing members with invalid or missing authentication."""
# Create a team and department first
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"]
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] != 0:
pytest.skip("Department creation failed, skipping auth test")
department_id: str = dept_res["data"]["id"]
# Try to list members with invalid auth
res: dict[str, Any] = list_department_members(invalid_auth, department_id)
assert res["code"] == expected_code, res
if expected_message:
assert expected_message in res["message"]
@pytest.mark.p1
class TestListMembers:
"""Comprehensive tests for listing department members."""
@pytest.fixture
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(web_api_auth, team_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_department(
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> dict[str, Any]:
"""Create a test department for use in tests."""
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": test_team["id"],
}
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_users(self, web_api_auth: RAGFlowWebApiAuth) -> list[dict[str, Any]]:
"""Create test users for use in tests."""
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"]})
return users
@pytest.fixture
def team_with_users(
self,
web_api_auth: RAGFlowWebApiAuth,
test_team: dict[str, Any],
test_users: list[dict[str, Any]],
) -> dict[str, Any]:
"""Add test users to the team."""
for user in test_users:
add_payload: dict[str, list[str]] = {"users": [user["email"]]}
add_users_to_team(web_api_auth, test_team["id"], add_payload)
return test_team
@pytest.fixture
def department_with_members(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
) -> dict[str, Any]:
"""Add test users to the department."""
if not test_users:
return test_department
user_ids: list[str] = [user["id"] for user in test_users]
add_payload: dict[str, list[str]] = {"user_ids": user_ids}
add_department_members(web_api_auth, test_department["id"], add_payload)
return test_department
@pytest.mark.p1
def test_list_members_empty_department(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test listing members from an empty department."""
department_id: str = test_department["id"]
res: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res["code"] == 0, res
assert "data" in res
assert isinstance(res["data"], list)
assert len(res["data"]) == 0
assert "message" in res
assert "0 member" in res["message"].lower()
@pytest.mark.p1
def test_list_members_with_multiple_users(
self,
web_api_auth: RAGFlowWebApiAuth,
department_with_members: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test listing members from a department with multiple users."""
if not test_users:
pytest.skip("No test users created")
department_id: str = department_with_members["id"]
res: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res["code"] == 0, res
assert "data" in res
assert isinstance(res["data"], list)
assert len(res["data"]) == len(test_users)
# Verify member structure
for member in res["data"]:
assert "user_id" in member
assert "email" in member
assert "nickname" in member
assert "status" in member
assert member["status"] == "1" # VALID status
# Verify all test users are in the list
member_user_ids: set[str] = {member["user_id"] for member in res["data"]}
test_user_ids: set[str] = {user["id"] for user in test_users}
assert member_user_ids == test_user_ids
@pytest.mark.p1
def test_list_members_invalid_department_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test listing members from a non-existent department."""
invalid_id: str = f"invalid_{uuid.uuid4().hex[:8]}"
res: dict[str, Any] = list_department_members(web_api_auth, invalid_id)
assert res["code"] == 100 # DATA_ERROR
assert "not found" in res["message"].lower()
@pytest.mark.p1
def test_list_members_not_team_member(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test listing members when user is not a team member."""
# Create a new user who is not in the team
email: str = f"testuser_{uuid.uuid4().hex[:8]}@example.com"
user_payload: dict[str, str] = {
"email": email,
"password": "TestPassword123!",
"nickname": "Test User",
}
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
if user_res["code"] != 0:
pytest.skip("User creation failed")
# Login as the new user (not in team)
new_user_auth: RAGFlowWebApiAuth = login_as_user(email, "TestPassword123!")
# Try to list members (user is not a team member)
res: dict[str, Any] = list_department_members(new_user_auth, test_department["id"])
assert res["code"] == 103 # PERMISSION_ERROR
assert "team member" in res["message"].lower() or "member of the team" in res["message"].lower()
@pytest.mark.p1
def test_list_members_response_structure(
self,
web_api_auth: RAGFlowWebApiAuth,
department_with_members: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test that listing members returns the expected response structure."""
if not test_users:
pytest.skip("No test users created")
department_id: str = department_with_members["id"]
res: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res["code"] == 0
assert "data" in res
assert isinstance(res["data"], list)
assert "message" in res
assert isinstance(res["message"], str)
# Verify member object structure
if len(res["data"]) > 0:
member: dict[str, Any] = res["data"][0]
required_fields: set[str] = {
"id",
"user_id",
"status",
"nickname",
"email",
"avatar",
"is_active",
}
for field in required_fields:
assert field in member, f"Missing field: {field}"
@pytest.mark.p1
def test_list_members_after_adding(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test listing members after adding users to the department."""
if not test_users:
pytest.skip("No test users created")
department_id: str = test_department["id"]
# List members before adding
res_before: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res_before["code"] == 0
initial_count: int = len(res_before["data"])
# Add a user
user_id: str = test_users[0]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_res: dict[str, Any] = add_department_members(web_api_auth, department_id, add_payload)
assert add_res["code"] == 0
# List members after adding
res_after: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res_after["code"] == 0
assert len(res_after["data"]) == initial_count + 1
# Verify the added user is in the list
member_user_ids: set[str] = {member["user_id"] for member in res_after["data"]}
assert user_id in member_user_ids
@pytest.mark.p1
def test_list_members_only_valid_status(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test that only members with valid status are returned."""
if not test_users:
pytest.skip("No test users created")
department_id: str = test_department["id"]
# Add users to department
user_ids: list[str] = [user["id"] for user in test_users]
add_payload: dict[str, list[str]] = {"user_ids": user_ids}
add_department_members(web_api_auth, department_id, add_payload)
# List members - all should have valid status
res: dict[str, Any] = list_department_members(web_api_auth, department_id)
assert res["code"] == 0
for member in res["data"]:
assert member["status"] == "1" # VALID status
@pytest.mark.p1
def test_list_members_team_member_not_department_member(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test that a team member (not department member) can list department members."""
if not test_users:
pytest.skip("No test users created")
department_id: str = test_department["id"]
# Add one user to department
user_id: str = test_users[0]["id"]
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_department_members(web_api_auth, department_id, add_payload)
# Login as another team member (not in department)
other_user_email: str = test_users[1]["email"]
other_user_auth: RAGFlowWebApiAuth = login_as_user(other_user_email, "TestPassword123!")
# Team member should be able to list members
res: dict[str, Any] = list_department_members(other_user_auth, department_id)
assert res["code"] == 0
assert len(res["data"]) == 1
assert res["data"][0]["user_id"] == user_id
@pytest.mark.p2
def test_list_members_empty_string_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test listing members with empty string department ID."""
res: dict[str, Any] = list_department_members(web_api_auth, "")
assert res["code"] != 0
assert "not found" in res["message"].lower() or res["code"] == 100
@pytest.mark.p2
def test_list_members_special_characters_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test listing members with special characters in department ID."""
invalid_id: str = "dept-123_!@#$%"
res: dict[str, Any] = list_department_members(web_api_auth, invalid_id)
assert res["code"] != 0
assert "not found" in res["message"].lower() or res["code"] == 100
@pytest.mark.p2
def test_list_members_multiple_departments(
self,
web_api_auth: RAGFlowWebApiAuth,
test_team: dict[str, Any],
test_users: list[dict[str, Any]],
) -> None:
"""Test listing members from multiple departments."""
if not test_users:
pytest.skip("No test users created")
# Create multiple departments
departments = []
for i in range(2):
dept_name: str = f"Department {i} {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": test_team["id"],
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] == 0:
departments.append(dept_res["data"])
if len(departments) < 2:
pytest.skip("Department creation failed")
# Add different users to each department
dept1_add_payload: dict[str, list[str]] = {"user_ids": [test_users[0]["id"]]}
add_department_members(web_api_auth, departments[0]["id"], dept1_add_payload)
if len(test_users) > 1:
dept2_add_payload: dict[str, list[str]] = {"user_ids": [test_users[1]["id"]]}
add_department_members(web_api_auth, departments[1]["id"], dept2_add_payload)
# List members from each department
res1: dict[str, Any] = list_department_members(web_api_auth, departments[0]["id"])
assert res1["code"] == 0
assert len(res1["data"]) == 1
assert res1["data"][0]["user_id"] == test_users[0]["id"]
if len(test_users) > 1:
res2: dict[str, Any] = list_department_members(web_api_auth, departments[1]["id"])
assert res2["code"] == 0
assert len(res2["data"]) == 1
assert res2["data"][0]["user_id"] == test_users[1]["id"]
@pytest.mark.p2
def test_list_members_large_department(
self,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
team_with_users: dict[str, Any],
) -> None:
"""Test listing members from a department with many users."""
# Create multiple users
users = []
for i in range(10):
email = f"bulkuser{i}_{uuid.uuid4().hex[:8]}@example.com"
user_payload: dict[str, str] = {
"email": email,
"password": "TestPassword123!",
"nickname": f"Bulk 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) < 5:
pytest.skip("Not enough users created")
# Add users to team
for user in users:
add_payload: dict[str, list[str]] = {"users": [user["email"]]}
add_users_to_team(web_api_auth, test_department["tenant_id"], add_payload)
# Add users to department
user_ids: list[str] = [user["id"] for user in users]
dept_add_payload: dict[str, list[str]] = {"user_ids": user_ids}
add_department_members(web_api_auth, test_department["id"], dept_add_payload)
# List members
res: dict[str, Any] = list_department_members(web_api_auth, test_department["id"])
assert res["code"] == 0
assert len(res["data"]) == len(users)
# Verify all users are in the list
member_user_ids: set[str] = {member["user_id"] for member in res["data"]}
test_user_ids: set[str] = {user["id"] for user in users}
assert member_user_ids == test_user_ids

138
test/testcases/test_http_api/test_department_management/test_remove_member.py
View file

@ -26,6 +26,7 @@ from common import (
create_department,
create_team,
create_user,
login_as_user,
remove_department_member,
)
from configs import INVALID_API_TOKEN
@ -53,13 +54,13 @@ class TestAuthorization:
invalid_auth: RAGFlowWebApiAuth | None,
expected_code: int,
expected_message: str,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
) -> None:
"""Test removing members with invalid or missing authentication."""
# Create a team, department, and add a user
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(WebApiAuth, team_payload)
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")
@ -70,7 +71,7 @@ class TestAuthorization:
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] != 0:
pytest.skip("Department creation failed, skipping auth test")
@ -83,7 +84,7 @@ class TestAuthorization:
"password": "TestPassword123!",
"nickname": "Test User",
}
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:
pytest.skip("User creation failed, skipping auth test")
@ -91,11 +92,11 @@ class TestAuthorization:
# Add user to team
add_team_payload: dict[str, list[str]] = {"users": [email]}
add_users_to_team(WebApiAuth, tenant_id, add_team_payload)
add_users_to_team(web_api_auth, tenant_id, add_team_payload)
# Add user to department
add_dept_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_department_members(WebApiAuth, department_id, add_dept_payload)
add_department_members(web_api_auth, department_id, add_dept_payload)
# Try to remove member with invalid auth
res: dict[str, Any] = remove_department_member(
@ -111,30 +112,30 @@ class TestRemoveMember:
"""Comprehensive tests for removing members from a department."""
@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_department(
self, WebApiAuth: RAGFlowWebApiAuth, test_team: dict[str, Any]
self, web_api_auth: RAGFlowWebApiAuth, test_team: dict[str, Any]
) -> dict[str, Any]:
"""Create a test department for use in tests."""
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": test_team["id"],
}
res: dict[str, Any] = create_department(WebApiAuth, dept_payload)
res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert res["code"] == 0
return res["data"]
@pytest.fixture
def test_user_with_member(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_team: dict[str, Any],
test_department: dict[str, Any],
) -> dict[str, Any]:
@ -145,19 +146,19 @@ class TestRemoveMember:
"password": "TestPassword123!",
"nickname": "Test User",
}
user_res: dict[str, Any] = create_user(WebApiAuth, user_payload)
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
assert user_res["code"] == 0
user_id: str = user_res["data"]["id"]
# Add user to team
add_team_payload: dict[str, list[str]] = {"users": [email]}
add_users_to_team(WebApiAuth, test_team["id"], add_team_payload)
add_users_to_team(web_api_auth, test_team["id"], add_team_payload)
# Add user to department
add_dept_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_dept_payload
web_api_auth, test_department["id"], add_dept_payload
)
assert add_res["code"] == 0
assert len(add_res["data"]["added"]) == 1
@ -167,14 +168,14 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_single_member(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_user_with_member: dict[str, Any],
) -> None:
"""Test removing a single member from a department."""
user_id: str = test_user_with_member["id"]
res: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert res["code"] == 0, res
@ -187,13 +188,13 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_member_invalid_department_id(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_user_with_member: dict[str, Any],
) -> None:
"""Test removing a member from a non-existent department."""
user_id: str = test_user_with_member["id"]
res: dict[str, Any] = remove_department_member(
WebApiAuth, "non_existent_department_id_12345", user_id
web_api_auth, "non_existent_department_id_12345", user_id
)
assert res["code"] == 102
@ -204,7 +205,7 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_member_user_not_in_department(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_team: dict[str, Any],
) -> None:
@ -216,18 +217,18 @@ class TestRemoveMember:
"password": "TestPassword123!",
"nickname": "Not In Dept User",
}
user_res: dict[str, Any] = create_user(WebApiAuth, user_payload)
user_res: dict[str, Any] = create_user(web_api_auth, user_payload)
assert user_res["code"] == 0
user_id: str = user_res["data"]["id"]
# Add user to team but not department
add_team_payload: dict[str, list[str]] = {"users": [email]}
add_users_to_team(WebApiAuth, test_team["id"], add_team_payload)
add_users_to_team(web_api_auth, test_team["id"], add_team_payload)
# Try to remove from department
res: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert res["code"] == 102
@ -238,12 +239,12 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_member_invalid_user_id(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
) -> None:
"""Test removing a non-existent user from a department."""
res: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], "non_existent_user_id_12345"
web_api_auth, test_department["id"], "non_existent_user_id_12345"
)
# The API checks if user is in department first, so this should return not found
@ -255,7 +256,7 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_member_twice(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_user_with_member: dict[str, Any],
) -> None:
@ -264,14 +265,14 @@ class TestRemoveMember:
# Remove first time
res1: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert res1["code"] == 0
# Try to remove again - API is idempotent, so it succeeds again
# (the record exists but is soft-deleted, and we update it again)
res2: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert res2["code"] == 0 # API allows removing twice (idempotent)
assert "removed" in res2["message"].lower() or "success" in res2[
@ -281,14 +282,14 @@ class TestRemoveMember:
@pytest.mark.p1
def test_remove_member_response_structure(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_user_with_member: dict[str, Any],
) -> None:
"""Test that remove member returns the expected response structure."""
user_id: str = test_user_with_member["id"]
res: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert res["code"] == 0
@ -301,18 +302,81 @@ class TestRemoveMember:
@pytest.mark.p2
def test_remove_member_not_team_owner_or_admin(
self, WebApiAuth: RAGFlowWebApiAuth
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test removing members when user is not team owner or admin."""
# This test would require creating a team with a different user
# and then trying to remove members as a non-admin user
# For now, we'll skip this as it requires multi-user setup
pytest.skip("Requires multi-user setup to test permission restrictions")
# Create a team (current user is owner)
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, "Failed to create team"
tenant_id: str = team_res["data"]["id"]
# Create a department
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, "Failed to create department"
department_id: str = dept_res["data"]["id"]
# Create a normal user (not admin/owner)
normal_user_email: str = f"normaluser_{uuid.uuid4().hex[:8]}@example.com"
normal_user_password: str = "TestPassword123!"
normal_user_payload: dict[str, str] = {
"email": normal_user_email,
"password": normal_user_password,
"nickname": "Normal User",
}
normal_user_res: dict[str, Any] = create_user(web_api_auth, normal_user_payload)
assert normal_user_res["code"] == 0, "Failed to create normal user"
# Add the normal user to the team as a normal member (not admin)
add_team_payload: dict[str, list[str]] = {"users": [normal_user_email]}
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"
# 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!"
another_user_payload: dict[str, str] = {
"email": another_user_email,
"password": another_user_password,
"nickname": "Another User",
}
another_user_res: dict[str, Any] = create_user(web_api_auth, another_user_payload)
assert another_user_res["code"] == 0, "Failed to create another user"
another_user_id: str = another_user_res["data"]["id"]
# Add another user to the team
add_another_team_payload: dict[str, list[str]] = {"users": [another_user_email]}
add_another_team_res: dict[str, Any] = add_users_to_team(
web_api_auth, tenant_id, add_another_team_payload
)
assert add_another_team_res["code"] == 0, "Failed to add another user to team"
# 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(
web_api_auth, department_id, add_dept_payload
)
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)
# Try to remove a member as the normal user (should fail)
remove_res: dict[str, Any] = remove_department_member(
normal_user_auth, department_id, another_user_id
)
assert remove_res["code"] == 108, f"Expected permission error (108), got {remove_res}"
assert "only team owners or admins" in remove_res["message"].lower()
@pytest.mark.p2
def test_remove_and_re_add_member(
self,
WebApiAuth: RAGFlowWebApiAuth,
web_api_auth: RAGFlowWebApiAuth,
test_department: dict[str, Any],
test_user_with_member: dict[str, Any],
) -> None:
@ -321,14 +385,14 @@ class TestRemoveMember:
# Remove member
remove_res: dict[str, Any] = remove_department_member(
WebApiAuth, test_department["id"], user_id
web_api_auth, test_department["id"], user_id
)
assert remove_res["code"] == 0
# Add member back
add_payload: dict[str, list[str]] = {"user_ids": [user_id]}
add_res: dict[str, Any] = add_department_members(
WebApiAuth, test_department["id"], add_payload
web_api_auth, test_department["id"], add_payload
)
assert add_res["code"] == 0
assert len(add_res["data"]["added"]) == 1

548
test/testcases/test_http_api/test_department_management/test_update_department.py Normal file
View file

@ -0,0 +1,548 @@
#
# 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 uuid
from typing import Any
import pytest
from common import (
add_department_members,
add_users_to_team,
create_department,
create_team,
create_user,
get_user_info,
login_as_user,
update_department,
)
from configs import INVALID_API_TOKEN
from libs.auth import RAGFlowWebApiAuth
# ---------------------------------------------------------------------------
# Test Classes
# ---------------------------------------------------------------------------
@pytest.mark.p1
class TestAuthorization:
"""Tests for authentication behavior when updating departments."""
@pytest.mark.parametrize(
("invalid_auth", "expected_code", "expected_message"),
[
(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 updating department with invalid or missing authentication."""
# Create a team and department first
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"]
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
if dept_res["code"] != 0:
pytest.skip("Department creation failed, skipping auth test")
department_id: str = dept_res["data"]["id"]
# Try to update department with invalid auth
update_payload: dict[str, str] = {"name": "Updated Name"}
res: dict[str, Any] = update_department(invalid_auth, department_id, update_payload)
assert res["code"] == expected_code, res
if expected_message:
assert expected_message in res["message"]
@pytest.mark.p1
class TestUpdateDepartment:
"""Comprehensive tests for department update API."""
def _add_creator_as_member(
self, web_api_auth: RAGFlowWebApiAuth, department_id: str
) -> None:
"""Helper to add the current user as a department member."""
user_info: dict[str, Any] = get_user_info(web_api_auth)
assert user_info["code"] == 0, user_info
current_user_id: str = user_info["data"]["id"]
add_member_payload: dict[str, list[str]] = {"user_ids": [current_user_id]}
add_member_res: dict[str, Any] = add_department_members(
web_api_auth, department_id, add_member_payload
)
assert add_member_res["code"] == 0, add_member_res
@pytest.mark.p1
def test_update_department_name(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating a department's name."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update the department name
new_name: str = f"Updated Department {uuid.uuid4().hex[:8]}"
update_payload: dict[str, str] = {"name": new_name}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert "data" in update_res
assert update_res["data"]["name"] == new_name
assert update_res["data"]["id"] == department_id
assert update_res["data"]["tenant_id"] == tenant_id
@pytest.mark.p1
def test_update_department_description(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating a department's description."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update the department description
new_description: str = "This is an updated description"
update_payload: dict[str, str] = {"description": new_description}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert "data" in update_res
assert update_res["data"]["description"] == new_description
@pytest.mark.p1
def test_update_department_name_and_description(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating both name and description at once."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update both name and description
new_name: str = f"Updated Department {uuid.uuid4().hex[:8]}"
new_description: str = "Updated description"
update_payload: dict[str, str] = {
"name": new_name,
"description": new_description,
}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert "data" in update_res
assert update_res["data"]["name"] == new_name
assert update_res["data"]["description"] == new_description
@pytest.mark.p1
def test_update_department_empty_description(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test setting description to empty string (should set to None)."""
# 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 a department with description
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
"description": "Original description",
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update description to empty string
update_payload: dict[str, str] = {"description": ""}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert "data" in update_res
# Empty description should be converted to None
assert update_res["data"]["description"] is None or update_res["data"]["description"] == ""
@pytest.mark.p2
def test_update_department_invalid_department_id(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating a non-existent department."""
invalid_id: str = "invalid_department_id_12345"
update_payload: dict[str, str] = {"name": "Updated Name"}
res: dict[str, Any] = update_department(
web_api_auth, invalid_id, update_payload
)
assert res["code"] != 0, "Should fail for invalid department ID"
assert "not found" in res["message"].lower() or "Department not found" in res["message"]
@pytest.mark.p2
def test_update_department_empty_name(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating department with empty name (should fail)."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Try to update with empty name
update_payload: dict[str, str] = {"name": ""}
res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert res["code"] != 0, "Should fail for empty name"
assert "empty" in res["message"].lower() or "cannot be empty" in res["message"]
@pytest.mark.p2
def test_update_department_name_too_long(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating department with name exceeding 128 characters."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Try to update with name > 128 characters
long_name: str = "a" * 129
update_payload: dict[str, str] = {"name": long_name}
res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert res["code"] != 0, "Should fail for name > 128 characters"
assert "128" in res["message"] or "too long" in res["message"].lower()
@pytest.mark.p2
def test_update_department_no_fields(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating department without providing any fields."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Try to update without any fields
update_payload: dict[str, Any] = {}
res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert res["code"] != 0, "Should fail when no fields provided"
assert "no fields" in res["message"].lower() or "provide" in res["message"].lower()
@pytest.mark.p2
def test_update_department_not_member(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating a department when user is not a member."""
# Create a team (current user is owner)
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, "Failed to create team"
tenant_id: str = team_res["data"]["id"]
# Create a department
dept_payload: dict[str, str] = {
"name": f"Test Department {uuid.uuid4().hex[:8]}",
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, "Failed to create department"
department_id: str = dept_res["data"]["id"]
# Create another user
another_user_email: str = f"anotheruser_{uuid.uuid4().hex[:8]}@example.com"
another_user_password: str = "TestPassword123!"
another_user_payload: dict[str, str] = {
"email": another_user_email,
"password": another_user_password,
"nickname": "Another User",
}
another_user_res: dict[str, Any] = create_user(web_api_auth, another_user_payload)
assert another_user_res["code"] == 0, "Failed to create another user"
# Add another user to the team (but not to the department)
add_team_payload: dict[str, list[str]] = {"users": [another_user_email]}
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 another user to team"
# Login as the other user
another_user_auth: RAGFlowWebApiAuth = login_as_user(
another_user_email, another_user_password
)
# Try to update the department as a non-member (should fail)
update_payload: dict[str, str] = {"name": "Updated Name"}
res: dict[str, Any] = update_department(
another_user_auth, department_id, update_payload
)
assert res["code"] != 0, "Should fail when user is not a department member"
assert "member" in res["message"].lower() or "permission" in res["message"].lower()
@pytest.mark.p2
def test_update_department_response_structure(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test that update response has the correct structure."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update the department
new_name: str = f"Updated Department {uuid.uuid4().hex[:8]}"
update_payload: dict[str, str] = {"name": new_name}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert "data" in update_res
assert "message" in update_res
assert isinstance(update_res["data"], dict)
assert "id" in update_res["data"]
assert "name" in update_res["data"]
assert "tenant_id" in update_res["data"]
assert update_res["data"]["id"] == department_id
assert update_res["data"]["name"] == new_name
@pytest.mark.p2
def test_update_department_multiple_times(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating a department multiple times in sequence."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# First update
first_name: str = f"First Update {uuid.uuid4().hex[:8]}"
first_payload: dict[str, str] = {"name": first_name}
first_res: dict[str, Any] = update_department(
web_api_auth, department_id, first_payload
)
assert first_res["code"] == 0, first_res
assert first_res["data"]["name"] == first_name
# Second update
second_name: str = f"Second Update {uuid.uuid4().hex[:8]}"
second_payload: dict[str, str] = {"name": second_name}
second_res: dict[str, Any] = update_department(
web_api_auth, department_id, second_payload
)
assert second_res["code"] == 0, second_res
assert second_res["data"]["name"] == second_name
# Third update
third_name: str = f"Third Update {uuid.uuid4().hex[:8]}"
third_payload: dict[str, str] = {"name": third_name}
third_res: dict[str, Any] = update_department(
web_api_auth, department_id, third_payload
)
assert third_res["code"] == 0, third_res
assert third_res["data"]["name"] == third_name
@pytest.mark.p2
def test_update_department_only_description(
self, web_api_auth: RAGFlowWebApiAuth
) -> None:
"""Test updating only description without name."""
# 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 a department
dept_name: str = f"Test Department {uuid.uuid4().hex[:8]}"
dept_payload: dict[str, str] = {
"name": dept_name,
"tenant_id": tenant_id,
}
dept_res: dict[str, Any] = create_department(web_api_auth, dept_payload)
assert dept_res["code"] == 0, dept_res
department_id: str = dept_res["data"]["id"]
original_name: str = dept_res["data"]["name"]
# Add creator as department member
self._add_creator_as_member(web_api_auth, department_id)
# Update only description
new_description: str = "Only description updated"
update_payload: dict[str, str] = {"description": new_description}
update_res: dict[str, Any] = update_department(
web_api_auth, department_id, update_payload
)
assert update_res["code"] == 0, update_res
assert update_res["data"]["description"] == new_description
# Name should remain unchanged
assert update_res["data"]["name"] == original_name