Adding teams API

.devcontainer/post_create.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 # This file is run from the .vscode folder
 WORKSPACE_FOLDER=/workspaces/jira
+git config --global --add safe.directory /workspaces/jira
 
 # Start the Jira Server docker instance first so can be running while we initialise everything else
 # Need to ensure this --version matches what is in CI

jira/client.py

@@ -85,6 +85,7 @@
     Sprint,
     Status,
     StatusCategory,
+    Team,
     User,
     Version,
     Votes,
@@ -1309,6 +1310,192 @@ def update_filter(
         raw_filter_json = json.loads(r.text)
         return Filter(self._options, self._session, raw=raw_filter_json)
 
+    # Teams
+
+    def create_team(
+        self,
+        org_id: str,
+        description: str,
+        display_name: str,
+        team_type: str,
+        site_id: str = None,
+    ) -> Team:
+        """Creates a team, and adds the requesting user as the initial member.
+
+        Args:
+            org_id (str): organization identifier
+            description (str): description field of the team to be created
+            display_name (str): name of the team to be created
+            team_type (str): either 'OPEN' or 'MEMBER_INVITE'
+            site_id (Optional[str])
+
+        Returns:
+            Team
+        """
+        url = f"gateway/api/public/teams/v1/org/{org_id}/teams/"
+        payload = {
+            "description": description,
+            "displayName": display_name,
+            "teamType": team_type,
+        }
+        if site_id is not None:
+            payload["siteId"] = site_id
+        r = self._session.post(url, data=json.dumps(payload))
+        raw_team_json: dict[str, Any] = json_loads(r)
+        return Team(self._options, self._session, raw=raw_team_json)
+
+    def get_team(self, org_id: str, team_id: str, site_id: str = None) -> Team:
+        """Get the specified team.
+
+        Args:
+            org_id (str): organization identifier
+            team_id (str): team identifier
+            site_id (Optional[str])
+
+        Returns:
+            Team
+        """
+        url = f"gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}"
+        params = {}
+        if site_id is not None:
+            params = {"siteId": site_id}
+        r = self._session.get(url, params=params)
+        raw_team_json: dict[str, Any] = json_loads(r)
+        return Team(self._options, self._session, raw=raw_team_json)
+
+    def remove_team(
+        self,
+        org_id: str,
+        team_id: str,
+    ):
+        """Delete the specified team.
+
+        Args:
+            org_id (str): organization identifier
+            team_id (str): team identifier
+
+        Returns:
+            bool
+        """
+        url = f"gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}"
+        r = self._session.delete(url)
+        return r.ok
+
+    def update_team(
+        self,
+        org_id: str,
+        team_id: str,
+        description: str,
+        displayName: str,
+    ) -> Team:
+        """Modifies the specified team with new values.
+
+        Args:
+            org_id (str): organization identifier
+            team_id (str): team identifier
+
+        Returns:
+            Team
+        """
+        url = f"gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}"
+
+        headers = {"Accept": "application/json", "Content-Type": "application/json"}
+
+        payload = {}
+        if description != "":
+            payload["description"] = description
+        if displayName != "":
+            payload["displayName"] = displayName
+
+        response = self._session.request(
+            "PATCH", url, data=json.dumps(payload), headers=headers
+        )
+        raw_team_json: dict[str, Any] = json_loads(response)
+        return Team(self._options, self._session, raw=raw_team_json)
+
+    def _fetch_paginated(self, url, payload):
+        result_response = self._session.get(url, data=json.dumps(payload)).json()
+        has_next_page = result_response["pageInfo"]["hasNextPage"]
+        end_index = result_response["pageInfo"]["endCursor"]
+
+        while has_next_page:
+            payload["after"] = end_index
+            r2 = self._session.get(url, data=json.dumps(payload)).json()
+            for res in r2["results"]:
+                result_response["results"].append(res)
+            end_index = r2["pageInfo"]["endCursor"]
+            has_next_page = r2["pageInfo"]["hasNextPage"]
+        return result_response
+
+    def team_members(
+        self,
+        org_id: str,
+        team_id: str,
+    ) -> list[str]:
+        """Return the list of account Ids corresponding to the team members.
+
+        Args:
+            org_id (str): Id of the org.
+            team_id (str): Id of the team.
+
+        Returns:
+            list[str]
+        """
+        url = f"/gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}/members"
+        payload = {"first": 50}
+        r = self._fetch_paginated(url, payload)
+        result = []
+        for accounts in r["results"]:
+            result.append(accounts.get("accountId"))
+        return result
+
+    def add_team_members(
+        self,
+        org_id: str,
+        team_id: str,
+        members: list[str],
+    ) -> tuple[list[str], list[str]]:
+        """Adds a list of members (accountIds) to the team members.
+
+        Args:
+            org_id (str): Id of the org.
+            team_id (str): Id of the team.
+            members (list[str]): Account Ids of the new members.
+
+        Returns:
+            (list[str], list[str]): (list of successful addition, list of failure)
+        """
+        url = f"/gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}/members/add"
+        payload_members_list = [{"accountId": accountId} for accountId in members]
+        payload = {"members": payload_members_list}
+        r = self._session.post(url, data=json.dumps(payload))
+        response_json = r.json()
+        return response_json["members"], response_json["errors"]
+
+    def remove_team_members(
+        self,
+        org_id: str,
+        team_id: str,
+        members: list[str],
+    ) -> bool:
+        """Removes the specified members from the team.
+
+        Args:
+            team_id (str): Id of the team.
+            org_id (str): Id of the org.
+            members (list[str]): Account Ids of the new members.
+
+        Returns:
+            bool
+        """
+        url = (
+            f"/gateway/api/public/teams/v1/org/{org_id}/teams/{team_id}/members/remove"
+        )
+        payload_members_list = [{"accountId": accountId} for accountId in members]
+        payload = {"members": payload_members_list}
+        r = self._session.post(url, data=json.dumps(payload))
+        return r.ok
+
     # Groups
 
     def group(self, id: str, expand: Any = None) -> Group:
@@ -2719,7 +2906,7 @@ def request_types(self, service_desk: ServiceDesk) -> list[RequestType]:
             service_desk = service_desk.id
         url = (
             self.server_url
-            + f"/rest/servicedeskapi/servicedesk/{service_desk}/requesttype"
+            + "/rest/servicedeskapi/servicedesk/{service_desk}/requesttype"
         )
         headers = {"X-ExperimentalApi": "opt-in"}
         r_json = json_loads(self._session.get(url, headers=headers))

jira/resources.py

@@ -57,6 +57,7 @@ class AnyLike:
     "Resolution",
     "SecurityLevel",
     "Status",
+    "Team",
     "User",
     "Group",
     "CustomFieldOption",
@@ -1234,6 +1235,25 @@ def __init__(
         self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
 
 
+class Team(Resource):
+    """A Jira team."""
+
+    TEAM_API_BASE_URL = "{server}/gateway/api/public/teams/v1/"
+
+    def __init__(
+        self,
+        options: dict[str, str],
+        session: ResilientSession,
+        raw: dict[str, Any] = None,
+    ):
+        Resource.__init__(
+            self, "org/{0}/teams/{1}", options, session, base_url=self.TEAM_API_BASE_URL
+        )
+        if raw:
+            self._parse_raw(raw)
+        self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
+
+
 class Group(Resource):
     """A Jira user group."""
 
@@ -1521,6 +1541,7 @@ def dict2resource(
     # Agile specific resources
     r"sprints/[^/]+$": Sprint,
     r"views/[^/]+$": Board,
+    r"org\?(accountId)/teams\?(accountId).+$": Team,
 }
 
 

tests/resources/test_teams.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+
+from tests.conftest import JiraTestCase, allow_on_cloud
+
+
+@allow_on_cloud
+class TeamsTests(JiraTestCase):
+    def setUp(self):
+        JiraTestCase.setUp(self)
+        self.test_team_name = f"testTeamFor_{self.test_manager.project_a}"
+        self.test_team_type = "OPEN"
+        self.org_id = os.environ["CI_JIRA_ORG_ID"]
+        self.test_team_description = "test Description"
+
+    @contextmanager
+    def make_team(self, **kwargs):
+        try:
+            new_team = self.jira.create_team(
+                self.org_id,
+                self.test_team_description,
+                self.test_team_name,
+                self.test_team_type,
+            )
+
+            if len(kwargs):
+                raise ValueError("Incorrect kwarg used !")
+            yield new_team
+        finally:
+            new_team.delete()
+
+    def test_team_creation(self):
+        with self.make_team() as test_team:
+            self.assertEqual(
+                self.test_team_name,
+                test_team["displayName"],
+            )
+            self.assertEqual(self.test_team_description, test_team["description"])
+            self.assertEqual(self.test_team_type, test_team["teamType"])
+
+    def test_team_get(self):
+        with self.make_team() as test_team:
+            fetched_team = self.jira.get_team(self.org_id, test_team.id)
+            self.assertEqual(
+                self.test_team_name,
+                fetched_team["displayName"],
+            )
+
+    def test_team_deletion(self):
+        with self.make_team() as test_team:
+            ok = self.jira.remove_team(self.org_id, test_team.id)
+            self.assertTrue(ok)
+
+    def test_updating_team(self):
+        new_desc = "Fake new description"
+        new_name = "Fake new Name"
+        with self.make_team() as test_team:
+            updated_team = self.jira.update_team(
+                self.org_id, test_team.id, description=new_desc, displayName=new_name
+            )
+            self.assertEqual(new_name, updated_team["displayName"])
+            self.assertEqual(new_desc, updated_team["description"])
+
+    def test_adding_team_members(self):
+        with self.make_team() as test_team:
+            self.jira.add_team_members(
+                self.org_id, test_team.id, members=[self.user_admin["accountId"]]
+            )
+
+    def test_get_team_members(self):
+        expected_accounts_id = [self.user_admin["accountId"]]
+        with self.make_team() as test_team:
+            self.jira.add_team_members(
+                self.org_id, test_team.id, members=expected_accounts_id
+            )
+
+            fetched_account_ids = self.jira.team_members(self.org_id, test_team.id)
+            self.assertEqual(expected_accounts_id, fetched_account_ids)