blob: 19a38bfcac9025ef031fffed180e7279dca5f460 [file] [log] [blame]
"""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