Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_aslist_jinja(self):
self.assertEqual(
config.aslist("work, jira, {{jirastatus|lower|replace(' ','_')}}"),
['work', 'jira', "{{jirastatus|lower|replace(' ','_')}}"]
)
if self.config.get('use_https', default=True, to_type=asbool):
self.scheme = 'https'
else:
self.scheme = 'http'
self.verify_ssl = self.config.get(
'verify_ssl', default=True, to_type=asbool
)
self.membership = self.config.get('membership', False)
self.owned = self.config.get('owned', False)
self.exclude_repos = self.config.get('exclude_repos', [], aslist)
self.include_repos = self.config.get('include_repos', [], aslist)
self.exclude_regex = self.config.get('exclude_regex', None)
self.include_regex = self.config.get('include_regex', None)
self.include_repos = list(map(self.add_default_namespace, self.include_repos))
self.exclude_repos = list(map(self.add_default_namespace, self.exclude_repos))
self.include_regex = re.compile(self.include_regex) if self.include_regex else None
self.exclude_regex = re.compile(self.exclude_regex) if self.exclude_regex else None
self.import_labels_as_tags = self.config.get(
'import_labels_as_tags', default=False, to_type=asbool
)
self.label_template = self.config.get(
'label_template', default='{{label}}', to_type=six.text_type
)
self.filter_merge_requests = self.config.get(
'filter_merge_requests', default=False, to_type=asbool
self.auth = (host, token)
if self.config.get('use_https', default=True, to_type=asbool):
self.scheme = 'https'
else:
self.scheme = 'http'
self.verify_ssl = self.config.get(
'verify_ssl', default=True, to_type=asbool
)
self.membership = self.config.get('membership', False)
self.owned = self.config.get('owned', False)
self.exclude_repos = self.config.get('exclude_repos', [], aslist)
self.include_repos = self.config.get('include_repos', [], aslist)
self.exclude_regex = self.config.get('exclude_regex', None)
self.include_regex = self.config.get('include_regex', None)
self.include_repos = list(map(self.add_default_namespace, self.include_repos))
self.exclude_repos = list(map(self.add_default_namespace, self.exclude_repos))
self.include_regex = re.compile(self.include_regex) if self.include_regex else None
self.exclude_regex = re.compile(self.exclude_regex) if self.exclude_regex else None
self.import_labels_as_tags = self.config.get(
'import_labels_as_tags', default=False, to_type=asbool
)
self.label_template = self.config.get(
'label_template', default='{{label}}', to_type=six.text_type
)
self.filter_merge_requests = self.config.get(
self.main_section = main_section
self.main_config = main_config
self.target = target
self.desc_len = self._get_config_or_default('description_length', 35, asint);
self.anno_len = self._get_config_or_default('annotation_length', 45, asint);
self.inline_links = self._get_config_or_default('inline_links', True, asbool);
self.annotation_links = self._get_config_or_default('annotation_links', not self.inline_links, asbool)
self.annotation_comments = self._get_config_or_default('annotation_comments', True, asbool)
self.annotation_newlines = self._get_config_or_default('annotation_newlines', False, asbool)
self.shorten = self._get_config_or_default('shorten', False, asbool)
self.default_priority = self.config.get('default_priority', 'M')
self.add_tags = []
for raw_option in aslist(self.config.get('add_tags', '')):
option = raw_option.strip(' +;')
if option:
self.add_tags.append(option)
log.info("Working on [%s]", self.target)
def __init__(self, *args, **kw):
super(PhabricatorService, self).__init__(*args, **kw)
self.host = self.config.get("host", None)
# These read login credentials from ~/.arcrc
if self.host is not None:
self.api = phabricator.Phabricator(host=self.host)
else:
self.api = phabricator.Phabricator()
self.shown_user_phids = (
self.config.get("user_phids", None, aslist))
self.shown_project_phids = (
self.config.get("project_phids", None, aslist))
only_if_assigned = self.config.get('only_if_assigned', default=False,
to_type=lambda x: x not in [False, "False", ""])
self.ignore_cc = self.config.get('ignore_cc', default=only_if_assigned,
to_type=lambda x: x == "True")
self.ignore_author = self.config.get('ignore_author', default=only_if_assigned,
to_type=lambda x: x == "True")
self.ignore_owner = self.config.get('ignore_owner', default=False,
to_type=lambda x: x == "True")
self.ignore_reviewers = self.config.get('ignore_reviewers', default=False,
to_type=lambda x: x == "True")
def __init__(self, *args, **kw):
super(PhabricatorService, self).__init__(*args, **kw)
self.host = self.config.get("host", None)
# These read login credentials from ~/.arcrc
if self.host is not None:
self.api = phabricator.Phabricator(host=self.host)
else:
self.api = phabricator.Phabricator()
self.shown_user_phids = (
self.config.get("user_phids", None, aslist))
self.shown_project_phids = (
self.config.get("project_phids", None, aslist))
only_if_assigned = self.config.get('only_if_assigned', default=False,
to_type=lambda x: x not in [False, "False", ""])
self.ignore_cc = self.config.get('ignore_cc', default=only_if_assigned,
to_type=lambda x: x == "True")
self.ignore_author = self.config.get('ignore_author', default=only_if_assigned,
to_type=lambda x: x == "True")
self.ignore_owner = self.config.get('ignore_owner', default=False,
to_type=lambda x: x == "True")
def run_hooks(conf, name):
if conf.has_option('hooks', name):
pre_import = aslist(conf.get('hooks', name))
if pre_import is not None:
for hook in pre_import:
exit_code = subprocess.call(hook, shell=True)
if exit_code != 0:
msg = 'Non-zero exit code %d on hook %s' % (
exit_code, hook
)
log.error(msg)
raise RuntimeError(msg)
'password': password},
auth=auth['oauth']).json()
self.config.data.set('bitbucket_refresh_token',
response['refresh_token'])
auth['token'] = response['access_token']
self.requests_kwargs = {}
if 'token' in auth:
self.requests_kwargs['headers'] = {
'Authorization': 'Bearer ' + auth['token']}
elif 'basic' in auth:
self.requests_kwargs['auth'] = auth['basic']
self.exclude_repos = self.config.get('exclude_repos', [], aslist)
self.include_repos = self.config.get('include_repos', [], aslist)
self.filter_merge_requests = self.config.get(
'filter_merge_requests', default=False, to_type=asbool
)
self.project_owner_prefix = self.config.get(
'project_owner_prefix', default=False, to_type=asbool
)
def synchronize(issue_generator, conf, main_section, dry_run=False):
def _bool_option(section, option, default):
try:
return asbool(conf.get(section, option))
except (NoSectionError, NoOptionError):
return default
targets = aslist(conf.get(main_section, 'targets'))
services = set([conf.get(target, 'service') for target in targets])
key_list = build_key_list(services)
uda_list = build_uda_config_overrides(services)
if uda_list:
log.info(
'Service-defined UDAs exist: you can optionally use the '
'`bugwarrior-uda` command to export a list of UDAs you can '
'add to your taskrc file.'
)
static_fields = ['priority']
if conf.has_option(main_section, 'static_fields'):
static_fields = aslist(conf.get(main_section, 'static_fields'))
# Before running CRUD operations, call the pre_import hook(s).