ansible-later/ansiblelater/utils/rulehelper.py
2019-04-04 15:35:47 +02:00

135 lines
4.3 KiB
Python

import codecs
from collections import defaultdict
import yaml
from yamllint import linter
from yamllint.config import YamlLintConfig
from ansiblelater.command.review import Error
from ansiblelater.exceptions import LaterAnsibleError, LaterError
from .yamlhelper import (action_tasks, normalize_task, normalized_yaml,
parse_yaml_linenumbers)
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["ansible"]["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["ansible"]["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