Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
fmt="{value}") == " abcd "
# String flags supported in addition to numeric
assert utils.validate_regex(
"- abcd -", r'-(?P[^-]+)-', 'i', fmt="{value}") == "abcd"
assert utils.validate_regex(
"- abcd -", r'-(?P[^-]+)-', re.I, fmt="{value}") == "abcd"
# Test multiple flag settings
assert utils.validate_regex(
"- abcd -", r'-(?P[^-]+)-', 'isax', fmt="{value}") == "abcd"
# Invalid flags are just ignored. The below fails to match
# because the default value of 'i' is over-ridden by what is
# identfied below, and no flag is set at the end of the day
assert utils.validate_regex(
"- abcd -", r'-(?P[ABCD]+)-', '-%2gb', fmt="{value}") is None
assert utils.validate_regex(
"- abcd -", r'-(?P[ABCD]+)-', '', fmt="{value}") is None
assert utils.validate_regex(
"- abcd -", r'-(?P[ABCD]+)-', None, fmt="{value}") is None
def __init__(self, token_a, token_b, token_c, include_image=True,
**kwargs):
"""
Initialize Microsoft Teams Object
"""
super(NotifyMSTeams, self).__init__(**kwargs)
self.token_a = validate_regex(
token_a, *self.template_tokens['token_a']['regex'])
if not self.token_a:
msg = 'An invalid MSTeams (first) Token ' \
'({}) was specified.'.format(token_a)
self.logger.warning(msg)
raise TypeError(msg)
self.token_b = validate_regex(
token_b, *self.template_tokens['token_b']['regex'])
if not self.token_b:
msg = 'An invalid MSTeams (second) Token ' \
'({}) was specified.'.format(token_b)
self.logger.warning(msg)
raise TypeError(msg)
self.token_c = validate_regex(
def __init__(self, token, targets, include_image=False, **kwargs):
"""
Initialize Gitter Object
"""
super(NotifyGitter, self).__init__(**kwargs)
# Secret Key (associated with project)
self.token = validate_regex(
token, *self.template_tokens['token']['regex'])
if not self.token:
msg = 'An invalid Gitter API Token ' \
'({}) was specified.'.format(token)
self.logger.warning(msg)
raise TypeError(msg)
# Parse our targets
self.targets = parse_list(targets)
if not self.targets:
msg = 'There are no valid Gitter targets to notify.'
self.logger.warning(msg)
raise TypeError(msg)
# Used to track maping of rooms to their numeric id lookup for
# messaging
def __init__(self, apikey, providerkey=None, priority=None, **kwargs):
"""
Initialize Prowl Object
"""
super(NotifyProwl, self).__init__(**kwargs)
if priority not in PROWL_PRIORITIES:
self.priority = self.template_args['priority']['default']
else:
self.priority = priority
# API Key (associated with project)
self.apikey = validate_regex(
apikey, *self.template_tokens['apikey']['regex'])
if not self.apikey:
msg = 'An invalid Prowl API Key ' \
'({}) was specified.'.format(apikey)
self.logger.warning(msg)
raise TypeError(msg)
# Store the provider key (if specified)
if providerkey:
self.providerkey = validate_regex(
providerkey, *self.template_tokens['providerkey']['regex'])
if not self.providerkey:
msg = 'An invalid Prowl Provider Key ' \
'({}) was specified.'.format(providerkey)
self.logger.warning(msg)
raise TypeError(msg)
def __init__(self, secret_key, **kwargs):
"""
Initialize Pushjet Object
"""
super(NotifyPushjet, self).__init__(**kwargs)
# Secret Key (associated with project)
self.secret_key = validate_regex(secret_key)
if not self.secret_key:
msg = 'An invalid Pushjet Secret Key ' \
'({}) was specified.'.format(secret_key)
self.logger.warning(msg)
raise TypeError(msg)
return
"""
Initialize Ryver Object
"""
super(NotifyRyver, self).__init__(**kwargs)
# API Token (associated with project)
self.token = validate_regex(
token, *self.template_tokens['token']['regex'])
if not self.token:
msg = 'An invalid Ryver API Token ' \
'({}) was specified.'.format(token)
self.logger.warning(msg)
raise TypeError(msg)
# Organization (associated with project)
self.organization = validate_regex(
organization, *self.template_tokens['organization']['regex'])
if not self.organization:
msg = 'An invalid Ryver Organization ' \
'({}) was specified.'.format(organization)
self.logger.warning(msg)
raise TypeError(msg)
# Store our webhook mode
self.mode = None \
if not isinstance(mode, six.string_types) else mode.lower()
if self.mode not in RYVER_WEBHOOK_MODES:
msg = 'The Ryver webhook mode specified ({}) is invalid.' \
.format(mode)
self.logger.warning(msg)
raise TypeError(msg)
token_b, *self.template_tokens['token_b']['regex'])
if not self.token_b:
msg = 'An invalid Slack (second) Token ' \
'({}) was specified.'.format(token_b)
self.logger.warning(msg)
raise TypeError(msg)
self.token_c = validate_regex(
token_c, *self.template_tokens['token_c']['regex'])
if not self.token_c:
msg = 'An invalid Slack (third) Token ' \
'({}) was specified.'.format(token_c)
self.logger.warning(msg)
raise TypeError(msg)
else:
self.access_token = validate_regex(
access_token, *self.template_tokens['access_token']['regex'])
if not self.access_token:
msg = 'An invalid Slack OAuth Access Token ' \
'({}) was specified.'.format(access_token)
self.logger.warning(msg)
raise TypeError(msg)
if not self.user:
self.logger.warning(
'No user was specified; using "%s".' % self.app_id)
# Build list of channels
self.channels = parse_list(targets)
if len(self.channels) == 0:
# No problem; the webhook is smart enough to just notify the
# channel it was created for; adding 'None' is just used as
def __init__(self, apikey, secret, source, targets=None, ttl=None,
**kwargs):
"""
Initialize Nexmo Object
"""
super(NotifyNexmo, self).__init__(**kwargs)
# API Key (associated with project)
self.apikey = validate_regex(
apikey, *self.template_tokens['apikey']['regex'])
if not self.apikey:
msg = 'An invalid Nexmo API Key ' \
'({}) was specified.'.format(apikey)
self.logger.warning(msg)
raise TypeError(msg)
# API Secret (associated with project)
self.secret = validate_regex(
secret, *self.template_tokens['secret']['regex'])
if not self.secret:
msg = 'An invalid Nexmo API Secret ' \
'({}) was specified.'.format(secret)
self.logger.warning(msg)
raise TypeError(msg)
if not self.aws_access_key_id:
msg = 'An invalid AWS Access Key ID was specified.'
self.logger.warning(msg)
raise TypeError(msg)
# Store our AWS API Secret Access key
self.aws_secret_access_key = validate_regex(secret_access_key)
if not self.aws_secret_access_key:
msg = 'An invalid AWS Secret Access Key ' \
'({}) was specified.'.format(secret_access_key)
self.logger.warning(msg)
raise TypeError(msg)
# Acquire our AWS Region Name:
# eg. us-east-1, cn-north-1, us-west-2, ...
self.aws_region_name = validate_regex(
region_name, *self.template_tokens['region']['regex'])
if not self.aws_region_name:
msg = 'An invalid AWS Region ({}) was specified.'.format(
region_name)
self.logger.warning(msg)
raise TypeError(msg)
# Initialize topic list
self.topics = list()
# Initialize numbers list
self.phone = list()
# Set our notify_url based on our region
self.notify_url = 'https://sns.{}.amazonaws.com/'\
.format(self.aws_region_name)
if not match:
# let outer exception handle this
raise TypeError
# The organization
self.organization = match.group('org')
if match.group('hostname'):
self.hostname = match.group('hostname')
except (TypeError, AttributeError):
msg = 'The Zulip organization specified ({}) is invalid.'\
.format(organization)
self.logger.warning(msg)
raise TypeError(msg)
self.token = validate_regex(
token, *self.template_tokens['token']['regex'])
if not self.token:
msg = 'The Zulip token specified ({}) is invalid.'\
.format(token)
self.logger.warning(msg)
raise TypeError(msg)
self.targets = parse_list(targets)
if len(self.targets) == 0:
# No channels identified, use default
self.targets.append(self.default_notification_channel)