| """Offers access to Fairphone's JIRA instance.""" |
| import logging |
| |
| from jira import JIRA |
| from jira.exceptions import JIRAError |
| |
| __log__ = logging.getLogger(__name__) |
| |
| |
| class JiraFairphone(JIRA): |
| """Abstraction for Fairphone's JIRA instance.""" |
| |
| HOST = "https://fairphone.atlassian.net" |
| |
| def __init__(self, username, password, dry_run=True): |
| """Construct a Fairphone JIRA client instance. |
| |
| Args: |
| username: JIRA username to log in with. |
| password: JIRA password for authentication. |
| dry_run: Do not modify remote. |
| |
| """ |
| super().__init__(JiraFairphone.HOST, basic_auth=(username, password)) |
| self.dry_run = dry_run |
| __log__.info("Logged in to JIRA as %s", username) |
| |
| def execute(self, func, *args, **kwargs): |
| """JIRA executing wrapper for handling dry-runs. |
| |
| Args: |
| func: The function to call |
| *args: Arguments to pass to the function. |
| **kwargs: Keyword arguments to pass to the function. |
| Returns: |
| The return of the function called. |
| |
| """ |
| if not self.dry_run: |
| return func(*args, **kwargs) |
| |
| return None |
| |
| def add_group(self, groupname): |
| """Create a new group in JIRA. |
| |
| Args: |
| groupname: The name of the group you wish to create. |
| Returns: |
| True if successful, otherwise False. |
| |
| """ |
| result = False |
| |
| if not self.groups(query=groupname): |
| __log__.info("Adding group %s", groupname) |
| result = self.execute(super().add_group, groupname) |
| |
| else: |
| __log__.debug("Group %s already exists", groupname) |
| |
| return result |
| |
| # pylint: disable=R0913 |
| def add_user( # noqa: N803 |
| self, |
| username, |
| email, |
| directoryId=1, |
| password=None, |
| fullname=None, |
| notify=False, |
| active=True, |
| ignore_existing=True, |
| application_keys=None, |
| ): |
| """Create a new JIRA user. |
| |
| Existing users will be ignored by default. |
| |
| Args: |
| username: the username of the new user |
| email: email address of the new user |
| directoryId: the directory ID the new user should be a part of |
| password: Optional, the password for the new user |
| fullname: Optional, the full name of the new user |
| notify: Whether or not to send a notification to the new user |
| active: Whether or not to make the new user active upon creation |
| ignore_existing: Ignore existing users. |
| applicationKeys: Keys of products user should have access to |
| Returns: |
| True if user was successfully added. |
| Throws: |
| JIRAError if super().add_user fails. |
| """ |
| result = False |
| |
| if self.search_users(username) and ignore_existing: |
| __log__.debug("User %s already exists", username) |
| |
| else: |
| __log__.info("Adding user %s [%s]", username, email) |
| result = self.execute( |
| super().add_user, |
| username, |
| email, |
| directoryId=directoryId, |
| password=password, |
| fullname=fullname, |
| notify=notify, |
| active=active, |
| ignore_existing=ignore_existing, |
| ) |
| |
| return result |
| |
| def get_jira_user_name(self, name): |
| """Get a JIRA user's user name. |
| |
| Args: |
| name: The string to use when searching. |
| Returns: |
| If found, the user name of the JIRA user . |
| |
| """ |
| user_name = None |
| |
| jira_users = self.search_users(name) |
| if not jira_users: |
| __log__.debug("Unable to find JIRA user %s", name) |
| |
| elif len(jira_users) == 1: |
| user_name = jira_users[0].name |
| __log__.debug("JIRA user name found %s", user_name) |
| |
| else: |
| __log__.debug( |
| "%d users found with name %s: %s", |
| len(jira_users), |
| name, |
| jira_users, |
| ) |
| |
| return user_name |
| |
| def add_user_to_group(self, username, group): |
| """Add a user to a group. |
| |
| If the group does not exist, create it. |
| |
| Args: |
| username: Username that will be added to specified group. |
| group: Group that the user will be added to. |
| Returns: |
| json response from JIRA server for success or False in case of |
| failure. |
| |
| """ |
| result = False |
| |
| jira_user_name = self.get_jira_user_name(username) |
| |
| if jira_user_name: |
| jira_group_members = {} |
| |
| if not self.groups(query=group): |
| __log__.warning( |
| "Unable to add user '%s' to '%s'", jira_user_name, group |
| ) |
| __log__.warning("Group %s is not a valid JIRA group", group) |
| return False |
| |
| jira_group_members = self.group_members(group) |
| |
| # Despite what the docs say add_user_to_group raises a JIRAException |
| # if the user already exists in a group. Let's check beforehand. |
| if jira_user_name not in jira_group_members: |
| __log__.info("Adding %s to group %s", jira_user_name, group) |
| result = self.execute( |
| super().add_user_to_group, jira_user_name, group |
| ) |
| |
| else: |
| __log__.debug("%s exists in group %s", jira_user_name, group) |
| |
| else: |
| __log__.warning( |
| "Unable to add unknown user '%s' to group '%s'", username, group |
| ) |
| |
| return result |
| |
| # pylint: disable=R0913 |
| def create_component( # noqa: N803 |
| self, |
| name, |
| project, |
| description=None, |
| leadUserName=None, |
| assigneeType=None, |
| isAssigneeTypeValid=False, |
| ): |
| """Create a component inside a project and return a Resource for it. |
| |
| Args: |
| name: name of the component |
| project: key of the project to create the component in |
| description: a description of the component |
| leadUserName: the username of the user responsible for this |
| component |
| assigneeType: see the ComponentBean.AssigneeType class for valid |
| values |
| isAssigneeTypeValid: boolean specifying whether the assignee type |
| is acceptable |
| Returns: |
| The Component object created for success or False in case of |
| failure. |
| |
| """ |
| result = False |
| |
| project_components = self.project_components(project) |
| |
| component_list = [component.name for component in project_components] |
| |
| if name not in component_list: |
| __log__.info("Adding component %s to %s:", name, project) |
| result = self.execute( |
| super().create_component, |
| name, |
| project, |
| description=description, |
| leadUserName=self.get_jira_user_name(leadUserName), |
| assigneeType="PROJECT_LEAD", |
| isAssigneeTypeValid=isAssigneeTypeValid, |
| ) |
| |
| else: |
| __log__.warning( |
| "Component %s already exists in project %s", name, project |
| ) |
| |
| return result |
| |
| # pylint: disable=C0103 |
| # Mirrors super class' leadUserName naming of the parameter |
| def update_component( # noqa: N803 |
| self, component_id, project, description=None, leadUserName=None |
| ): |
| """Update an existing component. |
| |
| Args: |
| component_id: ID of the component. |
| project: Name of the project the component lives in. |
| description: A description of the component. |
| leadUserName: The username of the user responsible for this |
| component. |
| Returns: |
| True if successful, otherwise False. |
| |
| """ |
| result = False |
| |
| try: |
| component = self.component(component_id) |
| fields = {} |
| |
| if leadUserName != getattr(component, "leadUserName", None): |
| fields["leadUserName"] = self.get_jira_user_name(leadUserName) |
| fields["assigneeType"] = "PROJECT_LEAD" |
| |
| if description != getattr(component, "description", None): |
| fields["description"] = description |
| |
| if fields: |
| __log__.info( |
| "%s %s component update: %s", project, component, fields |
| ) |
| |
| if not self.dry_run: |
| component.update(fields=fields) |
| result = True |
| |
| except JIRAError as error: |
| __log__.warning( |
| "Unable to find/update JIRA component %s in %s: %s", |
| component_id, |
| project, |
| error, |
| ) |
| |
| return result |