diff --git a/.devcontainer/post_create.sh b/.devcontainer/post_create.sh index 272b6ba23..424884396 100755 --- a/.devcontainer/post_create.sh +++ b/.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 diff --git a/jira/client.py b/jira/client.py old mode 100644 new mode 100755 index 1c8534869..421c90ff8 --- a/jira/client.py +++ b/jira/client.py @@ -71,6 +71,7 @@ IssueType, IssueTypeScheme, NotificationScheme, + Organization, PermissionScheme, Priority, PriorityScheme, @@ -85,6 +86,7 @@ Sprint, Status, StatusCategory, + Team, User, Version, Votes, @@ -1309,6 +1311,243 @@ def update_filter( raw_filter_json = json.loads(r.text) return Filter(self._options, self._session, raw=raw_filter_json) + # Organisations + def _get_service_desk_url(self) -> str: + """Returns the service desk root url. + + Returns: + str: service desk api url + """ + return f"{self.server_url}/rest/servicedeskapi" + + def create_org(self, org_name: str) -> Organization: + url = f"{self._get_service_desk_url()}/organization" + payload = {"name": org_name} + r = self._session.post(url, data=json.dumps(payload)) + raw_org_json: dict[str, Any] = json_loads(r) + return Organization(self._options, self._session, raw=raw_org_json) + + def remove_org(self, org_id: str) -> bool: + url = f"{self._get_service_desk_url()}/organization/{org_id}" + r = self._session.delete(url) + return r.ok + + def org(self, org_id: str) -> Organization: + url = f"{self._get_service_desk_url()}/organization/{org_id}" + r = self._session.get(url) + raw_org_json: dict[str, Any] = json_loads(r) + if r.status_code == 200: + return Organization(self._options, self._session, raw=raw_org_json) + return None + + def orgs(self, start=0, limit=50) -> ResultList[Organization]: + url = f"{self._get_service_desk_url()}/organization" + return self._fetch_pages( + Organization, "values", url, start, limit, base=self.server_url + ) + + def org_users(self, org_id, start=0, limit=50) -> ResultList[User]: + url = f"{self._get_service_desk_url()}/organization/{org_id}/user" + return self._fetch_pages(User, None, url, start, limit, base=self.server_url) + + def add_users_to_org(self, org_id: str, users: list[str]) -> bool: + url = f"{self._get_service_desk_url()}/organization/{org_id}/user" + payload = {"usernames": users} + r = self._session.post(url, data=json.dumps(payload)) + return r.ok + + def remove_users_from_org(self, org_id: str, users: list[str]) -> bool: + url = f"{self._get_service_desk_url()}/organization/{org_id}/user" + payload = {"usernames": users} + r = self._session.delete(url, data=json.dumps(payload)) + return r.ok + + # 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: @@ -1622,7 +1861,7 @@ def supports_service_desk(self): Returns: bool """ - url = self.server_url + "/rest/servicedeskapi/info" + url = f"{self._get_service_desk_url()}/info" headers = {"X-ExperimentalApi": "opt-in"} try: r = self._session.get(url, headers=headers) @@ -1640,7 +1879,7 @@ def create_customer(self, email: str, displayName: str) -> Customer: Returns: Customer """ - url = self.server_url + "/rest/servicedeskapi/customer" + url = f"{self._get_service_desk_url()}/customer" headers = {"X-ExperimentalApi": "opt-in"} r = self._session.post( url, @@ -1660,7 +1899,7 @@ def service_desks(self) -> list[ServiceDesk]: Returns: List[ServiceDesk] """ - url = self.server_url + "/rest/servicedeskapi/servicedesk" + url = f"{self._get_service_desk_url()}/servicedesk" headers = {"X-ExperimentalApi": "opt-in"} r_json = json_loads(self._session.get(url, headers=headers)) projects = [ @@ -1721,7 +1960,7 @@ def create_customer_request( elif isinstance(p, str): data["requestTypeId"] = self.request_type_by_name(service_desk, p).id - url = self.server_url + "/rest/servicedeskapi/request" + url = f"{self._get_service_desk_url()}/request" headers = {"X-ExperimentalApi": "opt-in"} r = self._session.post(url, headers=headers, data=json.dumps(data)) @@ -2717,10 +2956,7 @@ def request_types(self, service_desk: ServiceDesk) -> list[RequestType]: """ if hasattr(service_desk, "id"): service_desk = service_desk.id - url = ( - self.server_url - + f"/rest/servicedeskapi/servicedesk/{service_desk}/requesttype" - ) + url = f"{self._get_service_desk_url()}/servicedesk/{service_desk}/requesttype" headers = {"X-ExperimentalApi": "opt-in"} r_json = json_loads(self._session.get(url, headers=headers)) request_types = [ diff --git a/jira/resources.py b/jira/resources.py index 57ec31bbc..6dfee22fc 100644 --- a/jira/resources.py +++ b/jira/resources.py @@ -57,6 +57,8 @@ class AnyLike: "Resolution", "SecurityLevel", "Status", + "Organization", + "Team", "User", "Group", "CustomFieldOption", @@ -1234,6 +1236,46 @@ def __init__( self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw) +class Organization(Resource): + """A JIRA Organization.""" + + def __init__( + self, + options: dict[str, str], + session: ResilientSession, + raw: dict[str, Any] = None, + ): + Resource.__init__( + self, + "organization/{0}", + options, + session, + "{server}/rest/servicedeskapi/{path}", + ) + if raw: + self._parse_raw(raw) + 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 +1563,7 @@ def dict2resource( # Agile specific resources r"sprints/[^/]+$": Sprint, r"views/[^/]+$": Board, + r"org\?(accountId)/teams\?(accountId).+$": Team, } diff --git a/tests/resources/test_organisation.py b/tests/resources/test_organisation.py new file mode 100644 index 000000000..fe31855f8 --- /dev/null +++ b/tests/resources/test_organisation.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +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_org_name = "testOrg" + + @contextmanager + def make_org(self): + try: + new_org = self.jira.create_org(self.test_org_name) + yield new_org + finally: + new_org.delete() + + def test_org_creation(self): + with self.make_org() as test_org: + self.assertEqual(self.test_org_name, test_org["name"]) + + def test_org_deletion(self): + with self.make_org() as test_org: + ok = self.jira.remove_org(test_org.id) + self.assertTrue(ok) + + def test_fetch_one_org(self): + with self.make_org() as test_org: + found_org = self.jira.org(test_org.id) + self.assertEqual(test_org["name"], found_org["name"]) + + def test_fetch_multiple_orgs(self): + expected_org_was_found = False + with self.make_org() as test_org: + found_orgs = self.jira.orgs() + self.assertGreater(len(found_orgs), 0) + for org in found_orgs: + if org["name"] == test_org["name"]: + expected_org_was_found = True + self.assertTrue(expected_org_was_found) + + def test_add_users_to_org(self): + users_to_add = [self.user_admin.id] + with self.make_org() as test_org: + ok = self.jira.add_users_to_org(test_org.id, users_to_add) + self.assertTrue(ok) + + def test_get_users_in_org(self): + users_to_add = [self.user_admin.id] + with self.make_org() as test_org: + ok = self.jira.add_users_to_org(test_org.id, users_to_add) + self.assertTrue(ok) + found_users = self.jira.org_users(test_org.id) + self.assertIn(self.user_admin.id, found_users) + + def test_remove_user_from_org(self): + users_to_add = [self.user_admin.id] + with self.make_org() as test_org: + ok = self.jira.add_users_to_org(test_org.id, users_to_add) + self.assertTrue(ok) + found_users = self.jira.org_users(test_org.id) + self.assertIn(self.user_admin.id, found_users) + removal_ok = self.jira.remove_users_from_org(test_org.id, users_to_add) + self.assertTrue(removal_ok) + found_users_after_removal = self.jira.org_users(test_org.id) + self.assertNotIn(self.user_admin.id, found_users_after_removal) diff --git a/tests/resources/test_teams.py b/tests/resources/test_teams.py new file mode 100644 index 000000000..8bacbbdef --- /dev/null +++ b/tests/resources/test_teams.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from contextlib import contextmanager + +from tests.conftest import JiraTestCase, allow_on_cloud + + +@allow_on_cloud +class TeamsTests(JiraTestCase): + @classmethod + def setUpClass(cls): + JiraTestCase.setUp(cls) + cls.test_team_name = f"testTeamFor_{cls.test_manager.project_a}" + cls.test_team_type = "OPEN" + cls.org = cls.jira.create_org("TestOrgUsedByTeamsAPI") + cls.org_id = cls.org.id + cls.test_team_description = "test Description" + + @classmethod + def tearDownClass(cls): + cls.org.delete() + + @contextmanager + def make_team(self): + try: + new_team = self.jira.create_team( + self.org_id, + self.test_team_description, + self.test_team_name, + self.test_team_type, + ) + 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)