From 8eb24f7e287177ad55cd3ab84e93cf7a1bb438b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jorge=20Alberto=20D=C3=ADaz=20Orozco=20=28Akiel=29?= Date: Thu, 5 Dec 2024 20:59:31 +0100 Subject: [PATCH] Add more teams endpoints --- rocketchat_API/APISections/teams.py | 155 ++++++++++++++ tests/test_teams.py | 300 +++++++++++++++++++++++++++- 2 files changed, 446 insertions(+), 9 deletions(-) diff --git a/rocketchat_API/APISections/teams.py b/rocketchat_API/APISections/teams.py index 5b83132..39f6d92 100644 --- a/rocketchat_API/APISections/teams.py +++ b/rocketchat_API/APISections/teams.py @@ -1,5 +1,8 @@ +from rocketchat_API.APIExceptions.RocketExceptions import RocketMissingParamException from rocketchat_API.APISections.base import RocketChatBase +ID_OR_TEAM_NAME_REQUIRED = "team_id or team_name required" + class RocketChatTeams(RocketChatBase): def teams_create(self, name, team_type, **kwargs): @@ -7,3 +10,155 @@ def teams_create(self, name, team_type, **kwargs): return self.call_api_post( "teams.create", name=name, type=team_type, kwargs=kwargs ) + + def teams_delete(self, team_id=None, team_name=None, **kwargs): + """Delete a team.""" + if team_id: + return self.call_api_post("teams.delete", teamId=team_id, kwargs=kwargs) + if team_name: + return self.call_api_post("teams.delete", teamName=team_name, kwargs=kwargs) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_list_all(self, **kwargs): + """ + List all the teams on the server. + The calling user must have the 'view-all-teams' permission + """ + return self.call_api_get("teams.listAll", kwargs=kwargs) + + def teams_info(self, team_id=None, team_name=None, **kwargs): + """ + Gets a team's information. + If the team is not public, the caller user must be a member of + the team. + """ + if team_id: + return self.call_api_get("teams.info", teamId=team_id, kwargs=kwargs) + if team_name: + return self.call_api_get("teams.info", teamName=team_name, kwargs=kwargs) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_members( + self, team_id=None, team_name=None, name="", username="", **kwargs + ): + """Lists the members of a team.""" + if team_id: + return self.call_api_get( + "teams.members", + teamId=team_id, + name=name, + username=username, + kwargs=kwargs, + ) + if team_name: + return self.call_api_get( + "teams.members", + teamName=team_name, + name=name, + username=username, + kwargs=kwargs, + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_add_members(self, team_id=None, team_name=None, members=None, **kwargs): + """Adds members to the team.""" + if team_id: + return self.call_api_post( + "teams.addMembers", teamId=team_id, members=members, kwargs=kwargs + ) + if team_name: + return self.call_api_post( + "teams.addMembers", teamName=team_name, members=members, kwargs=kwargs + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_remove_member(self, team_id=None, team_name=None, user_id=None, **kwargs): + """Removes a member from a team. Requires edit-team-member permission.""" + if team_id: + return self.call_api_post( + "teams.removeMember", teamId=team_id, userId=user_id, kwargs=kwargs + ) + if team_name: + return self.call_api_post( + "teams.removeMember", teamName=team_name, userId=user_id, kwargs=kwargs + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_update_member(self, team_id=None, team_name=None, member=None, **kwargs): + """Updates a team member's roles. Requires edit-team-member permission.""" + if team_id: + return self.call_api_post( + "teams.updateMember", teamId=team_id, member=member, kwargs=kwargs + ) + if team_name: + return self.call_api_post( + "teams.updateMember", teamName=team_name, member=member, kwargs=kwargs + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_list_rooms( + self, team_id=None, team_name=None, room_type="", name="", **kwargs + ): + """List all rooms of the team.""" + if team_id: + return self.call_api_get( + "teams.listRooms", + teamId=team_id, + type=room_type, + filter=name, + kwargs=kwargs, + ) + if team_name: + return self.call_api_get( + "teams.listRooms", + teamName=team_name, + type=room_type, + filter=name, + kwargs=kwargs, + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_add_rooms(self, team_id=None, team_name=None, rooms=None, **kwargs): + """Adds rooms to the team. Requires add-team-channel permission.""" + if team_id: + return self.call_api_post( + "teams.addRooms", + teamId=team_id, + rooms=rooms, + kwargs=kwargs, + ) + if team_name: + return self.call_api_post( + "teams.addRooms", + teamName=team_name, + rooms=rooms, + kwargs=kwargs, + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_remove_room(self, team_id=None, team_name=None, room_id=None, **kwargs): + """Removes a room from a team. Requires remove-team-channel permission.""" + if team_id: + return self.call_api_post( + "teams.removeRoom", + teamId=team_id, + roomId=room_id, + kwargs=kwargs, + ) + if team_name: + return self.call_api_post( + "teams.removeRoom", + teamName=team_name, + roomId=room_id, + kwargs=kwargs, + ) + raise RocketMissingParamException(ID_OR_TEAM_NAME_REQUIRED) + + def teams_update_room(self, room_id, is_default, **kwargs): + """Updates a room from a team. Requires edit-team-channel permission.""" + return self.call_api_post( + "teams.updateRoom", + roomId=room_id, + isDefault=is_default, + kwargs=kwargs, + ) diff --git a/tests/test_teams.py b/tests/test_teams.py index c7057ff..a62113c 100644 --- a/tests/test_teams.py +++ b/tests/test_teams.py @@ -1,13 +1,295 @@ import uuid +import pytest -def test_team_create(logged_rocket): +from rocketchat_API.APIExceptions.RocketExceptions import RocketMissingParamException + + +@pytest.fixture +def test_team_name(): + return str(uuid.uuid1()) + + +@pytest.fixture +def test_team_id(test_team_name, logged_rocket): + _test_team_id = ( + logged_rocket.teams_create(name=test_team_name, team_type=1) + .json() + .get("team") + .get("_id") + ) + return _test_team_id + + +@pytest.fixture +def testuser_id(logged_rocket): + testuser = logged_rocket.users_info(username="testuser1").json() + + _testuser_id = testuser.get("user").get("_id") + + yield _testuser_id + + logged_rocket.users_delete(_testuser_id) + + +@pytest.fixture +def test_group_name(): + return str(uuid.uuid1()) + + +@pytest.fixture +def test_group_id(test_group_name, logged_rocket): + _test_group_id = ( + logged_rocket.groups_create(test_group_name).json().get("group").get("_id") + ) + return _test_group_id + + +def test_teams_create_delete(logged_rocket): name = str(uuid.uuid1()) - team_create = logged_rocket.teams_create( - name=name, - team_type=1, - ).json() - assert team_create.get("success") - assert "team" in team_create - assert team_create.get("team").get("name") == name - assert team_create.get("team").get("type") == 1 + teams_create = logged_rocket.teams_create(name=name, team_type=1).json() + assert teams_create.get("success") + assert name == teams_create.get("team").get("name") + teams_delete = logged_rocket.teams_delete(team_name=name).json() + assert teams_delete.get("success") + teams_create = logged_rocket.teams_create(name=name, team_type=1).json() + assert teams_create.get("success") + team_id = teams_create.get("team").get("_id") + teams_delete = logged_rocket.teams_delete(team_id=team_id).json() + assert teams_delete.get("success") + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_delete() + + +def test_teams_list_all(logged_rocket): + teams_list = logged_rocket.teams_list_all().json() + assert teams_list.get("success") + assert "teams" in teams_list + + +def test_teams_info(logged_rocket, test_team_name, test_team_id): + teams_info_by_id = logged_rocket.teams_info(team_id=test_team_id).json() + assert teams_info_by_id.get("success") + assert "teamInfo" in teams_info_by_id + assert teams_info_by_id.get("teamInfo").get("_id") == test_team_id + + teams_info_by_name = logged_rocket.teams_info(team_name=test_team_name).json() + assert teams_info_by_name.get("success") + assert "teamInfo" in teams_info_by_name + assert teams_info_by_name.get("teamInfo").get("_id") == test_team_id + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_info() + + +def test_teams_members(logged_rocket, test_team_name, test_team_id): + teams_members = logged_rocket.teams_members(team_id=test_team_id).json() + assert teams_members.get("success") + teams_members = logged_rocket.teams_members(team_name=test_team_name).json() + assert teams_members.get("success") + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_members() + + +def test_teams_add_update_remove_members(logged_rocket, test_team_id, testuser_id): + teams_add_members = logged_rocket.teams_add_members( + team_id=test_team_id, + members=[ + { + "userId": testuser_id, + "roles": [ + "member", + ], + } + ], + ).json() + assert teams_add_members.get("success"), teams_add_members.get("error") + + teams_members = logged_rocket.teams_members(team_id=test_team_id).json() + assert teams_members.get("success") + assert "members" in teams_members + assert len(teams_members.get("members")) == 2 + user_ids = [ + member.get("user").get("_id") for member in teams_members.get("members") + ] + assert testuser_id in user_ids + + # Make testuser owner + teams_update_member = logged_rocket.teams_update_member( + team_id=test_team_id, member={"userId": testuser_id, "roles": ["owner"]} + ).json() + assert teams_update_member.get("success"), teams_update_member.get("error") + + teams_members = logged_rocket.teams_members(team_id=test_team_id).json() + testuser_member = list( + filter( + lambda member: member.get("user").get("_id") == testuser_id, + teams_members.get("members"), + ) + )[0] + assert "owner" in testuser_member.get("roles") + + teams_remove_member = logged_rocket.teams_remove_member( + team_id=test_team_id, user_id=testuser_id + ).json() + assert teams_remove_member.get("success"), teams_remove_member.get("error") + teams_members = logged_rocket.teams_members(team_id=test_team_id).json() + assert "members" in teams_members + assert len(teams_members.get("members")) == 1 + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_add_members() + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_update_member() + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_remove_member() + + +def test_teams_add_update_remove_members_team_name( + logged_rocket, test_team_name, test_team_id, testuser_id +): + teams_add_members = logged_rocket.teams_add_members( + team_name=test_team_name, + members=[ + { + "userId": testuser_id, + "roles": [ + "member", + ], + } + ], + ).json() + assert teams_add_members.get("success"), teams_add_members.get("error") + + teams_members = logged_rocket.teams_members(team_name=test_team_name).json() + assert teams_members.get("success") + assert "members" in teams_members + assert len(teams_members.get("members")) == 2 + user_ids = [ + member.get("user").get("_id") for member in teams_members.get("members") + ] + assert testuser_id in user_ids + + # Make testuser owner + teams_update_member = logged_rocket.teams_update_member( + team_name=test_team_name, member={"userId": testuser_id, "roles": ["owner"]} + ).json() + assert teams_update_member.get("success"), teams_update_member.get("error") + + teams_members = logged_rocket.teams_members(team_name=test_team_name).json() + testuser_member = list( + filter( + lambda member: member.get("user").get("_id") == testuser_id, + teams_members.get("members"), + ) + )[0] + assert "owner" in testuser_member.get("roles") + + teams_remove_member = logged_rocket.teams_remove_member( + team_name=test_team_name, user_id=testuser_id + ).json() + assert teams_remove_member.get("success"), teams_remove_member.get("error") + teams_members = logged_rocket.teams_members(team_name=test_team_name).json() + assert "members" in teams_members + assert len(teams_members.get("members")) == 1 + + +def test_teams_list_rooms(logged_rocket, test_team_name, test_team_id): + teams_rooms = logged_rocket.teams_list_rooms( + team_id=test_team_id, room_type=1 + ).json() + assert teams_rooms.get("success") + assert "rooms" in teams_rooms + + teams_rooms_name = logged_rocket.teams_list_rooms( + team_name=test_team_name, room_type=1 + ).json() + assert teams_rooms_name.get("success") + assert "rooms" in teams_rooms_name + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_list_rooms() + + +def test_teams_add_update_remove_rooms(logged_rocket, test_team_id, test_group_id): + created_room = logged_rocket.teams_add_rooms( + team_id=test_team_id, rooms=[test_group_id] + ).json() + assert created_room.get("success"), created_room.get("error") + assert "rooms" in created_room + assert created_room.get("rooms")[0]["_id"] == test_group_id + + teams_rooms = logged_rocket.teams_list_rooms(team_id=test_team_id).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 1 + assert teams_rooms.get("rooms")[0]["_id"] == test_group_id + + teams_update_room = logged_rocket.teams_update_room( + test_group_id, is_default=True + ).json() + assert teams_update_room.get("success"), teams_update_room.get("success") + assert teams_update_room.get("room")["_id"] == test_group_id + assert teams_update_room.get("room")["teamDefault"] + + teams_rooms = logged_rocket.teams_list_rooms(team_id=test_team_id).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 1 + assert teams_rooms.get("rooms")[0]["_id"] == test_group_id + assert teams_rooms.get("rooms")[0]["teamDefault"] + + teams_remove_room = logged_rocket.teams_remove_room( + team_id=test_team_id, room_id=test_group_id + ).json() + assert teams_remove_room.get("success"), teams_remove_room.get("error") + + teams_rooms = logged_rocket.teams_list_rooms(team_id=test_team_id).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 0 + + +def test_teams_add_update_remove_rooms_name( + logged_rocket, test_team_name, test_team_id, test_group_id +): + created_room = logged_rocket.teams_add_rooms( + team_name=test_team_name, rooms=[test_group_id] + ).json() + assert created_room.get("success"), created_room.get("error") + assert "rooms" in created_room + assert created_room.get("rooms")[0]["_id"] == test_group_id + + teams_rooms = logged_rocket.teams_list_rooms(team_name=test_team_name).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 1 + assert teams_rooms.get("rooms")[0]["_id"] == test_group_id + + teams_update_room = logged_rocket.teams_update_room( + test_group_id, is_default=True + ).json() + assert teams_update_room.get("success"), teams_update_room.get("success") + assert teams_update_room.get("room")["_id"] == test_group_id + assert teams_update_room.get("room")["teamDefault"] + + teams_rooms = logged_rocket.teams_list_rooms(team_name=test_team_name).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 1 + assert teams_rooms.get("rooms")[0]["_id"] == test_group_id + assert teams_rooms.get("rooms")[0]["teamDefault"] + + teams_remove_room = logged_rocket.teams_remove_room( + team_name=test_team_name, room_id=test_group_id + ).json() + assert teams_remove_room.get("success"), teams_remove_room.get("error") + + teams_rooms = logged_rocket.teams_list_rooms(team_name=test_team_name).json() + assert teams_rooms.get("success"), teams_rooms.get("error") + assert len(teams_rooms.get("rooms")) == 0 + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_add_rooms() + + with pytest.raises(RocketMissingParamException): + logged_rocket.teams_remove_room()