From 5413628aaf4cd91622d685655f0d703ec9903088 Mon Sep 17 00:00:00 2001 From: Hetavi Shah Date: Mon, 17 Nov 2025 18:49:26 +0530 Subject: [PATCH] [OND211-2329]: Added tests to test the add/remove members to a department. --- test/testcases/test_http_api/common.py | 112 +++++ .../test_add_members.py | 414 ++++++++++++++++++ .../test_remove_member.py | 336 ++++++++++++++ 3 files changed, 862 insertions(+) create mode 100644 test/testcases/test_http_api/test_department_management/test_add_members.py create mode 100644 test/testcases/test_http_api/test_department_management/test_remove_member.py diff --git a/test/testcases/test_http_api/common.py b/test/testcases/test_http_api/common.py index bbe129f10..d67a47d0a 100644 --- a/test/testcases/test_http_api/common.py +++ b/test/testcases/test_http_api/common.py @@ -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() diff --git a/test/testcases/test_http_api/test_department_management/test_add_members.py b/test/testcases/test_http_api/test_department_management/test_add_members.py new file mode 100644 index 000000000..cebab79c7 --- /dev/null +++ b/test/testcases/test_http_api/test_department_management/test_add_members.py @@ -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() + diff --git a/test/testcases/test_http_api/test_department_management/test_remove_member.py b/test/testcases/test_http_api/test_department_management/test_remove_member.py new file mode 100644 index 000000000..4f4704414 --- /dev/null +++ b/test/testcases/test_http_api/test_department_management/test_remove_member.py @@ -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 +