[OND211-2329]: Added tests to test the add/remove members to a department.
This commit is contained in:
Hetavi Shah
parent
0e2c72a2bb
commit
5413628aaf
3 changed files with 862 additions and 0 deletions
|
|
@ -413,6 +413,62 @@ def create_team(
|
|||
return res.json()
|
||||
|
||||
|
||||
def add_users_to_team(
|
||||
auth: Union[AuthBase, str, None],
|
||||
tenant_id: str,
|
||||
payload: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Add users to a team (tenant).
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
tenant_id: The tenant/team ID to add users to.
|
||||
payload: Optional JSON payload containing users list (emails).
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing added and failed user lists.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{TEAM_API_URL}/{tenant_id}/users/add"
|
||||
res: requests.Response = requests.post(
|
||||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
def remove_users_from_team(
|
||||
auth: Union[AuthBase, str, None],
|
||||
tenant_id: str,
|
||||
payload: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Remove users from a team (tenant).
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
tenant_id: The tenant/team ID to remove users from.
|
||||
payload: Optional JSON payload containing user_ids list.
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing removal results.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{TEAM_API_URL}/{tenant_id}/users/remove"
|
||||
res: requests.Response = requests.post(
|
||||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
# DEPARTMENT MANAGEMENT
|
||||
DEPARTMENT_API_URL: str = f"/{VERSION}/department"
|
||||
|
||||
|
|
@ -441,3 +497,59 @@ def create_department(
|
|||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
def add_department_members(
|
||||
auth: Union[AuthBase, str, None],
|
||||
department_id: str,
|
||||
payload: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Add members to a department.
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
department_id: The department ID to add members to.
|
||||
payload: Optional JSON payload containing user_ids list.
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing added and failed user lists.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{DEPARTMENT_API_URL}/{department_id}/members/add"
|
||||
res: requests.Response = requests.post(
|
||||
url=url, headers=headers, auth=auth, json=payload
|
||||
)
|
||||
return res.json()
|
||||
|
||||
|
||||
def remove_department_member(
|
||||
auth: Union[AuthBase, str, None],
|
||||
department_id: str,
|
||||
user_id: str,
|
||||
*,
|
||||
headers: Dict[str, str] = HEADERS,
|
||||
) -> Dict[str, Any]:
|
||||
"""Remove a member from a department.
|
||||
|
||||
Args:
|
||||
auth: Authentication object (AuthBase subclass), token string, or None.
|
||||
department_id: The department ID to remove member from.
|
||||
user_id: The user ID to remove from the department.
|
||||
headers: Optional HTTP headers. Defaults to HEADERS.
|
||||
|
||||
Returns:
|
||||
JSON response as a dictionary containing the removal result.
|
||||
|
||||
Raises:
|
||||
requests.RequestException: If the HTTP request fails.
|
||||
"""
|
||||
url: str = f"{HOST_ADDRESS}{DEPARTMENT_API_URL}/{department_id}/members/{user_id}"
|
||||
res: requests.Response = requests.delete(
|
||||
url=url, headers=headers, auth=auth
|
||||
)
|
||||
return res.json()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,414 @@
|
|||
#
|
||||
# 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,
|
||||
)
|
||||
from configs import INVALID_API_TOKEN
|
||||
from libs.auth import RAGFlowWebApiAuth
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test Classes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestAuthorization:
|
||||
"""Tests for authentication behavior when adding members to 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,
|
||||
WebApiAuth: 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)
|
||||
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(WebApiAuth, dept_payload)
|
||||
if dept_res["code"] != 0:
|
||||
pytest.skip("Department creation failed, skipping auth test")
|
||||
|
||||
department_id: str = dept_res["data"]["id"]
|
||||
|
||||
# Try to add members with invalid auth
|
||||
add_payload: dict[str, list[str]] = {"user_ids": ["test_user_id"]}
|
||||
res: dict[str, Any] = add_department_members(invalid_auth, department_id, add_payload)
|
||||
assert res["code"] == expected_code, res
|
||||
if expected_message:
|
||||
assert expected_message in res["message"]
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestAddMembers:
|
||||
"""Comprehensive tests for adding members to a department."""
|
||||
|
||||
@pytest.fixture
|
||||
def test_team(self, WebApiAuth: 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)
|
||||
assert res["code"] == 0
|
||||
return res["data"]
|
||||
|
||||
@pytest.fixture
|
||||
def test_department(
|
||||
self, WebApiAuth: 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)
|
||||
assert res["code"] == 0
|
||||
return res["data"]
|
||||
|
||||
@pytest.fixture
|
||||
def test_users(self, WebApiAuth: 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(WebApiAuth, 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]],
|
||||
) -> 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)
|
||||
return test_team
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_single_member(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding a single member to a department."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert res["code"] == 0, res
|
||||
assert "data" in res
|
||||
assert "added" in res["data"]
|
||||
assert "failed" in res["data"]
|
||||
assert len(res["data"]["added"]) == 1
|
||||
assert res["data"]["added"][0] == user_id
|
||||
assert len(res["data"]["failed"]) == 0
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_multiple_members(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding multiple members to a department."""
|
||||
if len(test_users) < 2:
|
||||
pytest.skip("Need at least 2 test users")
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert res["code"] == 0, res
|
||||
assert len(res["data"]["added"]) == 2
|
||||
assert set(res["data"]["added"]) == set(user_ids)
|
||||
assert len(res["data"]["failed"]) == 0
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_missing_user_ids(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 101
|
||||
assert "user_ids" in res["message"].lower() or "required" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_empty_user_ids(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 101
|
||||
assert "non-empty array" in res["message"].lower() or "empty" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_invalid_user_id(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test adding a non-existent user."""
|
||||
add_payload: dict[str, list[str]] = {
|
||||
"user_ids": ["non_existent_user_id_12345"]
|
||||
}
|
||||
res: dict[str, Any] = add_department_members(
|
||||
WebApiAuth, test_department["id"], add_payload
|
||||
)
|
||||
|
||||
assert res["code"] == 0 # API returns success with failed list
|
||||
assert "data" in res
|
||||
assert len(res["data"]["added"]) == 0
|
||||
assert len(res["data"]["failed"]) == 1
|
||||
assert "not found" in res["data"]["failed"][0]["error"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_user_not_in_team(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding a user who is not a member of the team."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
# Create a user but don't add them to the team
|
||||
email = f"notinteam_{uuid.uuid4().hex[:8]}@example.com"
|
||||
user_payload: dict[str, str] = {
|
||||
"email": email,
|
||||
"password": "TestPassword123!",
|
||||
"nickname": "Not In Team User",
|
||||
}
|
||||
user_res: dict[str, Any] = create_user(WebApiAuth, 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
|
||||
)
|
||||
|
||||
assert res["code"] == 0 # API returns success with failed list
|
||||
assert len(res["data"]["added"]) == 0
|
||||
assert len(res["data"]["failed"]) == 1
|
||||
assert "not a member of the team" in res["data"]["failed"][0]["error"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_duplicate_member(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding a user who is already in the department."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
user_id: str = test_users[0]["id"]
|
||||
|
||||
# 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
|
||||
)
|
||||
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
|
||||
)
|
||||
assert res2["code"] == 0 # API returns success with failed list
|
||||
assert len(res2["data"]["added"]) == 0
|
||||
assert len(res2["data"]["failed"]) == 1
|
||||
assert "already a member" in res2["data"]["failed"][0]["error"].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_invalid_department_id(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding members to a non-existent department."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert res["code"] == 102
|
||||
assert "department not found" in res["message"].lower() or "not found" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_invalid_user_id_format(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 0 # API returns success with failed list
|
||||
assert len(res["data"]["added"]) == 0
|
||||
assert len(res["data"]["failed"]) >= 1
|
||||
# All invalid formats should be in failed list
|
||||
for failed in res["data"]["failed"]:
|
||||
assert "invalid" in failed["error"].lower() or "format" in failed[
|
||||
"error"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_add_member_mixed_valid_invalid(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test adding a mix of valid and invalid user IDs."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
valid_user_id: str = test_users[0]["id"]
|
||||
invalid_user_id: str = "non_existent_user_id_12345"
|
||||
add_payload: dict[str, list[str]] = {
|
||||
"user_ids": [valid_user_id, invalid_user_id]
|
||||
}
|
||||
res: dict[str, Any] = add_department_members(
|
||||
WebApiAuth, test_department["id"], add_payload
|
||||
)
|
||||
|
||||
assert res["code"] == 0
|
||||
assert len(res["data"]["added"]) == 1
|
||||
assert res["data"]["added"][0] == valid_user_id
|
||||
assert len(res["data"]["failed"]) == 1
|
||||
assert res["data"]["failed"][0]["user_id"] == invalid_user_id
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_add_member_not_team_owner_or_admin(
|
||||
self, WebApiAuth: 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")
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_add_member_response_structure(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
team_with_users: dict[str, Any],
|
||||
test_users: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Test that add members returns the expected response structure."""
|
||||
if not test_users:
|
||||
pytest.skip("No test users created")
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert res["code"] == 0
|
||||
assert "data" in res
|
||||
assert isinstance(res["data"], dict)
|
||||
assert "added" in res["data"]
|
||||
assert "failed" in res["data"]
|
||||
assert isinstance(res["data"]["added"], list)
|
||||
assert isinstance(res["data"]["failed"], list)
|
||||
assert "message" in res
|
||||
assert "added" in res["message"].lower() or "member" in res["message"].lower()
|
||||
|
||||
|
|
@ -0,0 +1,336 @@
|
|||
#
|
||||
# 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,
|
||||
remove_department_member,
|
||||
)
|
||||
from configs import INVALID_API_TOKEN
|
||||
from libs.auth import RAGFlowWebApiAuth
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test Classes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestAuthorization:
|
||||
"""Tests for authentication behavior when removing members from 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,
|
||||
WebApiAuth: 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)
|
||||
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(WebApiAuth, dept_payload)
|
||||
if dept_res["code"] != 0:
|
||||
pytest.skip("Department creation failed, skipping auth test")
|
||||
|
||||
department_id: str = dept_res["data"]["id"]
|
||||
|
||||
# Create and add a user to team and department
|
||||
email = 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(WebApiAuth, user_payload)
|
||||
if user_res["code"] != 0:
|
||||
pytest.skip("User creation failed, skipping auth test")
|
||||
|
||||
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, 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)
|
||||
|
||||
# Try to remove member with invalid auth
|
||||
res: dict[str, Any] = remove_department_member(
|
||||
invalid_auth, department_id, user_id
|
||||
)
|
||||
assert res["code"] == expected_code, res
|
||||
if expected_message:
|
||||
assert expected_message in res["message"]
|
||||
|
||||
|
||||
@pytest.mark.p1
|
||||
class TestRemoveMember:
|
||||
"""Comprehensive tests for removing members from a department."""
|
||||
|
||||
@pytest.fixture
|
||||
def test_team(self, WebApiAuth: 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)
|
||||
assert res["code"] == 0
|
||||
return res["data"]
|
||||
|
||||
@pytest.fixture
|
||||
def test_department(
|
||||
self, WebApiAuth: 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)
|
||||
assert res["code"] == 0
|
||||
return res["data"]
|
||||
|
||||
@pytest.fixture
|
||||
def test_user_with_member(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_team: dict[str, Any],
|
||||
test_department: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Create a test user and add them to team and department."""
|
||||
email = 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(WebApiAuth, 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 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
|
||||
)
|
||||
assert add_res["code"] == 0
|
||||
assert len(add_res["data"]["added"]) == 1
|
||||
|
||||
return {"id": user_id, "email": email}
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_single_member(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 0, res
|
||||
assert "data" in res
|
||||
assert res["data"] is True
|
||||
assert "removed" in res["message"].lower() or "success" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_member_invalid_department_id(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 102
|
||||
assert "department not found" in res["message"].lower() or "not found" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_member_user_not_in_department(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
test_team: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing a user who is not in the department."""
|
||||
# Create a user and add to team but not department
|
||||
email = f"notindept_{uuid.uuid4().hex[:8]}@example.com"
|
||||
user_payload: dict[str, str] = {
|
||||
"email": email,
|
||||
"password": "TestPassword123!",
|
||||
"nickname": "Not In Dept User",
|
||||
}
|
||||
user_res: dict[str, Any] = create_user(WebApiAuth, 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)
|
||||
|
||||
# Try to remove from department
|
||||
res: dict[str, Any] = remove_department_member(
|
||||
WebApiAuth, test_department["id"], user_id
|
||||
)
|
||||
|
||||
assert res["code"] == 102
|
||||
assert "not a member" in res["message"].lower() or "not found" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_member_invalid_user_id(
|
||||
self,
|
||||
WebApiAuth: 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"
|
||||
)
|
||||
|
||||
# The API checks if user is in department first, so this should return not found
|
||||
assert res["code"] == 102
|
||||
assert "not a member" in res["message"].lower() or "not found" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_member_twice(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
test_user_with_member: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing the same member twice (idempotent operation)."""
|
||||
user_id: str = test_user_with_member["id"]
|
||||
|
||||
# Remove first time
|
||||
res1: dict[str, Any] = remove_department_member(
|
||||
WebApiAuth, 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
|
||||
)
|
||||
assert res2["code"] == 0 # API allows removing twice (idempotent)
|
||||
assert "removed" in res2["message"].lower() or "success" in res2[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p1
|
||||
def test_remove_member_response_structure(
|
||||
self,
|
||||
WebApiAuth: 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
|
||||
)
|
||||
|
||||
assert res["code"] == 0
|
||||
assert "data" in res
|
||||
assert res["data"] is True
|
||||
assert "message" in res
|
||||
assert "removed" in res["message"].lower() or "success" in res[
|
||||
"message"
|
||||
].lower()
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_remove_member_not_team_owner_or_admin(
|
||||
self, WebApiAuth: 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")
|
||||
|
||||
@pytest.mark.p2
|
||||
def test_remove_and_re_add_member(
|
||||
self,
|
||||
WebApiAuth: RAGFlowWebApiAuth,
|
||||
test_department: dict[str, Any],
|
||||
test_user_with_member: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test removing a member and then adding them back."""
|
||||
user_id: str = test_user_with_member["id"]
|
||||
|
||||
# Remove member
|
||||
remove_res: dict[str, Any] = remove_department_member(
|
||||
WebApiAuth, 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
|
||||
)
|
||||
assert add_res["code"] == 0
|
||||
assert len(add_res["data"]["added"]) == 1
|
||||
assert add_res["data"]["added"][0] == user_id
|
||||
|
||||
Loading…
Add table
Reference in a new issue