# By: Riasat Ullah # This module contains methods to validate details of a task. from utils import helpers, times, var_names from validations import string_validator import configuration import datetime import pytz def validate_task_details(timestamp, start_date, title, timezone, task_time, repeat_on, text_msg, urgency_level, organization_id=None, created_by=None, instantiate=True, alert=True, integration_id=None, api_key_id=None, routing_id=None, related_task_id=None, task_status=None, notes=None, tags=None, dedup_key=None, service_id=None, service_policy_id=None, assignees=None, amended_urgency=None, next_alert_timestamp=None, status_update=None, subscribers=None, conference_bridge=None, impacted_business_services=None): ''' Validate the details of a task. The set of details of a task must have all these attributes in the correct format. :param timestamp: timestamp when the task was created :param organization_id: ID of the organization this task is for :param start_date: date the task should start from :param title: title of the task :param timezone: timezone the task is being created for :param task_time: time the task should trigger at in the format HH:MM :param created_by: (can be None) user_id of the user who created the task :param repeat_on: (list) of days the task should be repeated on :param text_msg: text message of the task :param urgency_level: urgency level of the task :param instantiate: (boolean) True if instances should be created for this task :param alert: (boolean) True if alerts should be generated from the instances created this task :param integration_id: ID of the integration that created the task :param api_key_id: ID of the API key that created the instance :param routing_id: (list) of conditional routing ID :param related_task_id: ID of any other task this task is related to. This will be useful for grouping/suppressing :param task_status: a status of the task - GROUPED, SUPPRESSED, etc :param notes: any notes that were added to the task :param tags: list of tags to help classify a task :param dedup_key: a key to stop de-duplication; if this key is present in active task instance, then instances will not be created for subsequent tasks containing the key :param service_id: ID of the service this task is for :param service_policy_id: policy ID the service is associated to :param assignees: (list of int) of policy ids of assignees (both users and groups) :param amended_urgency: (integer) if priority is not internally changed then this stays as None :param next_alert_timestamp: timestamp the alert should be re-triggered next :param status_update: (str) status update to add to the instance :param subscribers: (list of user_ids) subscribers to add to the instance :param conference_bridge: (dict) details of conference bridge to add to the instance :param impacted_business_services: (list of int) IDs of business services that will eb impacted by this instance ''' # validate date and timestamp assert isinstance(timestamp, datetime.datetime) if start_date is not None: helpers.is_date_str_or_date(start_date) # check for sql injections assert isinstance(title, str) and len(title) > 0 if text_msg is not None: assert isinstance(text_msg, str) assert timezone in pytz.all_timezones if isinstance(task_time, str): times.get_time_from_string(task_time) else: assert isinstance(task_time, datetime.time) assert urgency_level in configuration.allowed_urgency_levels # validate repeat days if repeat_on is not None: assert isinstance(repeat_on, list) and all(isinstance(x, int) and 0 <= x <= 6 for x in repeat_on) # check all optional values if organization_id is not None: assert isinstance(organization_id, int) if created_by is not None: assert isinstance(created_by, int) assert isinstance(instantiate, bool) assert isinstance(alert, bool) if integration_id is not None: assert isinstance(integration_id, int) if api_key_id is not None: assert isinstance(api_key_id, int) if routing_id is not None: helpers.get_int_list(routing_id) if related_task_id is not None: assert isinstance(related_task_id, int) if task_status is not None: assert task_status in configuration.allowed_task_statuses if notes is not None: assert string_validator.is_not_sql_injection(notes) if dedup_key is not None: assert isinstance(dedup_key, str) if tags is not None: assert isinstance(tags, list) for item in tags: assert isinstance(item, str) if amended_urgency is not None: assert amended_urgency in configuration.allowed_urgency_levels if next_alert_timestamp is not None: assert isinstance(next_alert_timestamp, datetime.datetime) if status_update is not None: assert isinstance(status_update, str) if subscribers is not None: assert isinstance(subscribers, list) for id_ in subscribers: assert isinstance(id_, int) if conference_bridge is not None: assert isinstance(conference_bridge, dict) assert set(conference_bridge.keys()) == {var_names.conference_phone, var_names.conference_url, var_names.additional_info} if impacted_business_services is not None: helpers.get_int_list(impacted_business_services) # make sure that at least one assignable policy can be obtained assert service_id is not None or assignees is not None if service_id is not None: assert isinstance(service_id, int) if service_policy_id is not None: assert isinstance(service_policy_id, int) if assignees is not None: assert isinstance(assignees, list) for id_ in assignees: assert isinstance(id_, int) def task_time_is_forward_looking(task_timezone, task_time): ''' Check if a task time is forward looking or not. :param task_timezone: (str) timezone of the task :param task_time: (datetime.time) time the task should trigger at :return: (boolean) True if it is; False otherwise ''' task_time = times.get_time_from_string(task_time) task_utc_time = times.region_to_utc_time(times.get_current_timestamp().replace( hour=task_time.hour, minute=task_time.minute, second=0, microsecond=0), task_timezone) if task_utc_time <= times.get_current_timestamp(): return False return True