# By: Riasat Ullah # This class represents a policy that already exists. # A Policy be associated to other policies through their policy id # as long as there is way to back track or form a loop. from objects.policy_level import PolicyLevel from utils import constants, times, var_names class Policy(object): def __init__(self, policy_id, organization_id, policy_name, policy_type, levels, reference_id=None, associated_services=None, tags=None): ''' This class represents a policy object. :param policy_id: policy id :param organization_id: organization id :param policy_name: name of the policy (spaces are allowed) :param policy_type: USER or GROUP :param levels: (list) of PolicyLevel objects :param reference_id: the reference ID of the policy :param associated_services: (list of list) -> [ [serv name, serv ref], ... ] :param tags: tags associated with the policy ''' self.policy_id = policy_id self.organization_id = organization_id self.policy_name = policy_name self.policy_type = policy_type self.levels = levels self.reference_id = reference_id self.associated_services = associated_services self.tags = tags @staticmethod def create_policy(details): ''' Creates new Policy object from policy info dict. :param details: (dict) of policy info :return: Policy object ''' policy_id = details[var_names.policy_id] policy_levels = [] for item in details[var_names.levels]: policy_levels.append(PolicyLevel.create_level(item)) pol_ref_id = details[var_names.policy_ref_id] if var_names.policy_ref_id in details else None return Policy(policy_id, details[var_names.organization_id], details[var_names.policy_name], details[var_names.policy_type], policy_levels, pol_ref_id, details[var_names.tags] if var_names.tags in details else None) def get_on_call(self, level_number, check_datetime=None): ''' Gets the current on-call assignees of the policy. :param level_number: (int) level number to check for; starts from 1 :param check_datetime: (datetime.datetime) date and time to check for :return: (list of tuples) -> [(user_id, display name, assignee policy id), ...] of users who are on call; it will send an empty list if there are none ''' check_level = None for level in self.levels: if level.level == level_number: check_level = level break on_call = [] if check_level is not None: on_call = check_level.get_on_call(check_datetime) return on_call def get_next_available_level_on_call(self, level_number, check_datetime=None): ''' Gets the current next available level of the policy and its on-call assignees. :param level_number: (int) level number to start checking from (level numbers start from 1) :param check_datetime: (datetime.datetime) date and time to check for :return: (list of tuples) -> [(user_id, display name, assignee policy id), ...] of users who are on call; it will send an empty list if there are none ''' all_level_on_calls = self.get_all_level_on_calls(check_datetime=check_datetime) if level_number <= len(all_level_on_calls): for i in range(level_number - 1, len(all_level_on_calls)): item_level = all_level_on_calls[i] if len(item_level) != 0: return i + 1, item_level return level_number, [] def get_all_level_on_calls(self, check_datetime=None): ''' Get the on_calls for each of the levels of the routine. :param check_datetime: (datetime.datetime) date on which to check for :return: list of (list of tuples) ordered by level number (ascending) -> [ [(user_id, display name, assignee policy id), ...], ...] ''' on_calls = [] for i in range(0, len(self.levels)): # we do i + 1 because levels start at 1; not 0 on_calls.append(self.get_on_call(i + 1, check_datetime)) return on_calls def get_all_assignees(self, skip_exceptions=True): ''' Get all the assignees of a policy; from all routines and all levels. :param skip_exceptions: (boolean) True if exceptions should be ignored; False otherwise :return: (list of tuples) -> [(policy_id, policy name), ...] of users/assignees ''' assignees = [] for level in self.levels: assignees += level.get_all_assignees(skip_exceptions) return assignees def has_level(self, level_number, check_datetime=times.get_current_timestamp()): ''' Checks if a policy has a certain level or not. :param level_number: (int) level number :param check_datetime: (datetime.datetime) timestamp to check on :return: (boolean) True if it has; False otherwise ''' on_call = self.get_on_call(level_number, check_datetime) if len(on_call) > 0: return True return False def get_level_minutes(self, level_number): ''' Get the wait minutes of a given policy level. If the requested level does not exist, then sends the wait minutes of the first level. :param level_number: (int) level number :return: (int) wait minutes ''' if level_number <= len(self.levels): for level in self.levels: if level.level == level_number: return level.minutes return self.levels[0].minutes def is_user_policy(self): ''' Checks if the policy is a user policy or not. :return: True if it is; False otherwise ''' if self.policy_type == constants.user_policy: return True return False def is_group_policy(self): ''' Checks if the policy is a group policy or not. :return: True if it is; False otherwise ''' if self.policy_type == constants.group_policy: return True return False def get_routine(self, routine_id): ''' Get a routine given its routine id. :param routine_id: the routine id :return: Routine object; None if routine does not exist ''' for level in self.levels: for routine in level.routines: if routine.routine_id == routine_id: return routine return None def equivalent_to(self, new_policy): ''' Checks if two Policy objects are the same or not. :param new_policy: the Policy to compare against :return: (boolean) True if they are; False otherwise ''' assert isinstance(new_policy, Policy) if self.to_dict() == new_policy.to_dict(): return True return False def to_dict(self, basic_info=False, with_on_call=False): ''' Gets the dict of the Policy object :param basic_info: specifies if only the routine id and name are required :param with_on_call: True if the current on-call info is also required :return: dict of Policy object ''' levels_list = [] for level in self.levels: levels_list.append(level.to_dict(basic_info)) data = {var_names.policy_id: self.policy_id, var_names.policy_name: self.policy_name, var_names.policy_type: self.policy_type, var_names.levels: levels_list, var_names.policy_ref_id: self.reference_id, var_names.associated_services: self.associated_services, var_names.tags: self.tags} if not basic_info: data[var_names.organization_id] = self.organization_id if with_on_call: all_on_calls = self.get_all_level_on_calls() if basic_info: adjusted_on_calls = [] for level in all_on_calls: adjusted_on_calls.append([(x[1], x[0]) for x in level]) all_on_calls = adjusted_on_calls data[var_names.on_call] = all_on_calls return data