ansible-later/ansiblelater/utils/rulehelper.py
2018-12-19 11:19:07 +01:00

136 lines
4.4 KiB
Python

import codecs
import yaml
from collections import defaultdict
from yamllint import linter
from yamllint.config import YamlLintConfig
# Workaround for import errors with ansble 2.1 and 2.3
from ansible.parsing.dataloader import DataLoader
from ansiblelater import Error
from .yamlhelper import normalize_task
from .yamlhelper import action_tasks
from .yamlhelper import parse_yaml_linenumbers
from .yamlhelper import normalized_yaml
from .exceptions import LaterError, LaterAnsibleError
def get_tasks(candidate, settings):
errors = []
try:
with codecs.open(candidate.path, mode='rb', encoding='utf-8') as f:
yamllines = parse_yaml_linenumbers(f, candidate.path)
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
except LaterAnsibleError as e:
errors.append(Error(e.line, "syntax error: %s" % (e.message)))
return yamllines, errors
def get_action_tasks(candidate, settings):
tasks = []
errors = []
try:
with codecs.open(candidate.path, mode='rb', encoding='utf-8') as f:
yamllines = parse_yaml_linenumbers(f, candidate.path)
if yamllines:
tasks = action_tasks(yamllines, candidate)
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
except LaterAnsibleError as e:
errors.append(Error(e.line, "syntax error: %s" % (e.message)))
return tasks, errors
def get_normalized_task(task, candidate, settings):
normalized = None
errors = []
try:
normalized = normalize_task(task, candidate.path, settings.custom_modules)
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
except LaterAnsibleError as e:
errors.append(Error(e.line, "syntax error: %s" % (e.message)))
return normalized, errors
def get_normalized_tasks(candidate, settings):
normalized = []
errors = []
try:
with codecs.open(candidate.path, mode='rb', encoding='utf-8') as f:
yamllines = parse_yaml_linenumbers(f, candidate.path)
if yamllines:
tasks = action_tasks(yamllines, candidate)
for task in tasks:
# An empty `tags` block causes `None` to be returned if
# the `or []` is not present - `task.get('tags', [])`
# does not suffice.
if 'skip_ansible_lint' in (task.get('tags') or []):
# No need to normalize_task if we are skipping it.
continue
normalized.append(normalize_task(task, candidate.path, settings.custom_modules))
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
except LaterAnsibleError as e:
errors.append(Error(e.line, "syntax error: %s" % (e.message)))
return normalized, errors
def get_normalized_yaml(candidate, settings, options=None):
errors = []
if not options:
options = defaultdict(dict)
options.update(remove_empty=True)
options.update(remove_markers=True)
try:
yamllines = normalized_yaml(candidate.path, options)
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
except LaterAnsibleError as e:
errors.append(Error(e.line, "syntax error: %s" % (e.message)))
return yamllines, errors
def get_raw_yaml(candidate, settings):
content = None
errors = []
try:
with codecs.open(candidate.path, mode='rb', encoding='utf-8') as f:
content = yaml.safe_load(f)
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
return content, errors
def run_yamllint(candidate, settings, options="extends: default"):
errors = []
try:
with codecs.open(candidate.path, mode='rb', encoding='utf-8') as f:
for problem in linter.run(f, YamlLintConfig(options)):
errors.append(Error(problem.line, problem.desc))
except LaterError as ex:
e = ex.original
errors.append(Error(e.problem_mark.line + 1, "syntax error: %s" % (e.problem)))
return errors