diff --git a/api/apps/department_app.py b/api/apps/department_app.py index 04524bdee..c40c099e9 100644 --- a/api/apps/department_app.py +++ b/api/apps/department_app.py @@ -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 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 index cebab79c7..792c6fb25 100644 --- 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 @@ -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 diff --git a/test/testcases/test_http_api/test_department_management/test_create_department.py b/test/testcases/test_http_api/test_department_management/test_create_department.py index 2694b8a80..16149632c 100644 --- a/test/testcases/test_http_api/test_department_management/test_create_department.py +++ b/test/testcases/test_http_api/test_department_management/test_create_department.py @@ -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) diff --git a/test/testcases/test_http_api/test_department_management/test_delete_department.py b/test/testcases/test_http_api/test_department_management/test_delete_department.py new file mode 100644 index 000000000..8721dffa8 --- /dev/null +++ b/test/testcases/test_http_api/test_department_management/test_delete_department.py @@ -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 + diff --git a/test/testcases/test_http_api/test_department_management/test_list_members.py b/test/testcases/test_http_api/test_department_management/test_list_members.py new file mode 100644 index 000000000..a40ba7ef2 --- /dev/null +++ b/test/testcases/test_http_api/test_department_management/test_list_members.py @@ -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 + 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 index 4f4704414..ba87dee22 100644 --- 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 @@ -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 diff --git a/test/testcases/test_http_api/test_department_management/test_update_department.py b/test/testcases/test_http_api/test_department_management/test_update_department.py new file mode 100644 index 000000000..20209dbbc --- /dev/null +++ b/test/testcases/test_http_api/test_department_management/test_update_department.py @@ -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 +