mirror of
https://github.com/thegeeklab/ansible-later.git
synced 2024-11-14 17:20:39 +00:00
136 lines
4.4 KiB
Python
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.command.review import Error
|
|
from .yamlhelper import normalize_task
|
|
from .yamlhelper import action_tasks
|
|
from .yamlhelper import parse_yaml_linenumbers
|
|
from .yamlhelper import normalized_yaml
|
|
from ansiblelater.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
|