github/ansible-later
0
0
Fork 0
mirror of https://github.com/thegeeklab/ansible-later.git synced 2024-11-22 04:40:42 +00:00
Code Issues Projects Releases Wiki Activity

fix some linting errors

Browse Source
This commit is contained in:
Robert Kaussow 2019-04-04 16:06:18 +02:00
parent 2b574970c2
commit 554d957af3
9 changed files with 86 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ansiblelater/__main__.py
View File

@ -1,4 +1,5 @@
#!/usr/bin/env python
"""Main program."""
import argparse
import logging
@ -8,6 +9,7 @@ from ansiblelater.command import base, candidates
def main():
"""Run main program."""
parser = argparse.ArgumentParser(
description="Validate ansible files against best pratice guideline")
parser.add_argument("-c", "--config", dest="config_file",

52
ansiblelater/data/standards.py
View File

@ -1,30 +1,26 @@
from ansiblelater.standard import Standard
from ansiblelater.rules.yamlfiles import check_yaml_empty_lines
from ansiblelater.rules.yamlfiles import check_yaml_indent
from ansiblelater.rules.yamlfiles import check_yaml_hyphens
from ansiblelater.rules.yamlfiles import check_yaml_document_start
from ansiblelater.rules.yamlfiles import check_yaml_colons
from ansiblelater.rules.yamlfiles import check_yaml_file
from ansiblelater.rules.yamlfiles import check_yaml_has_content
from ansiblelater.rules.yamlfiles import check_native_yaml
from ansiblelater.rules.ansiblefiles import (check_become_user,
check_braces_spaces,
check_command_has_changes,
check_command_instead_of_module,
check_compare_to_literal_bool,
check_empty_string_compare,
check_filter_separation,
check_install_use_latest,
check_literal_bool_format,
check_name_format,
check_named_task,
check_shell_instead_command,
check_unique_named_task)
from ansiblelater.rules.rolefiles import check_meta_main, check_scm_in_src
from ansiblelater.rules.taskfiles import check_line_between_tasks
from ansiblelater.rules.rolefiles import check_meta_main
from ansiblelater.rules.rolefiles import check_scm_in_src
from ansiblelater.rules.ansiblefiles import check_unique_named_task
from ansiblelater.rules.ansiblefiles import check_named_task
from ansiblelater.rules.ansiblefiles import check_name_format
from ansiblelater.rules.ansiblefiles import check_braces_spaces
from ansiblelater.rules.ansiblefiles import check_command_instead_of_module
from ansiblelater.rules.ansiblefiles import check_install_use_latest
from ansiblelater.rules.ansiblefiles import check_shell_instead_command
from ansiblelater.rules.ansiblefiles import check_command_has_changes
from ansiblelater.rules.ansiblefiles import check_empty_string_compare
from ansiblelater.rules.ansiblefiles import check_compare_to_literal_bool
from ansiblelater.rules.ansiblefiles import check_literal_bool_format
from ansiblelater.rules.ansiblefiles import check_become_user
from ansiblelater.rules.ansiblefiles import check_filter_separation
from ansiblelater.rules.yamlfiles import (check_native_yaml, check_yaml_colons,
check_yaml_document_start,
check_yaml_empty_lines,
check_yaml_file,
check_yaml_has_content,
check_yaml_hyphens,
check_yaml_indent)
from ansiblelater.standard import Standard
tasks_should_be_separated = Standard(dict(
id="ANSIBLE0001",
@ -225,8 +221,8 @@ use_yaml_rather_than_key_value = Standard(dict(
))
ansible_min_version = '2.1'
ansible_review_min_version = '0.1.0'
ansible_min_version = "2.5"
ansible_review_min_version = "0.1.0"
standards = [

1
ansiblelater/logger.py
View File

@ -76,6 +76,7 @@ def get_logger(name=None, level=logging.DEBUG, json=False):
def update_logger(logger, level=None, json=None):
"""Update logger configuration to change logging settings."""
for handler in logger.handlers[:]:
logger.removeHandler(handler)

80
ansiblelater/rules/ansiblefiles.py
View File

@ -1,8 +1,8 @@
import re
import os
import re
from collections import defaultdict
from ansiblelater.command.review import Result, Error
from ansiblelater.command.review import Error, Result
from ansiblelater.utils import count_spaces
from ansiblelater.utils.rulehelper import (get_normalized_tasks,
get_normalized_yaml)
@ -32,16 +32,16 @@ def check_braces_spaces(candidate, settings):
def check_named_task(candidate, settings):
tasks, errors = get_normalized_tasks(candidate, settings)
nameless_tasks = ['meta', 'debug', 'include_role', 'import_role',
'include_tasks', 'import_tasks', 'include_vars',
'block']
nameless_tasks = ["meta", "debug", "include_role", "import_role",
"include_tasks", "import_tasks", "include_vars",
"block"]
description = "module '%s' used without name attribute"
if not errors:
for task in tasks:
module = task["action"]["__ansible_module__"]
if 'name' not in task and module not in nameless_tasks:
errors.append(Error(task['__line__'], description % module))
if "name" not in task and module not in nameless_tasks:
errors.append(Error(task["__line__"], description % module))
return Result(candidate.path, errors)
@ -53,8 +53,8 @@ def check_name_format(candidate, settings):
if not errors:
for task in tasks:
if 'name' in task:
namelines[task['name']].append(task['__line__'])
if "name" in task:
namelines[task["name"]].append(task["__line__"])
for (name, lines) in namelines.items():
if not name[0].isupper():
errors.append(Error(lines[-1], description % name))
@ -70,8 +70,8 @@ def check_unique_named_task(candidate, settings):
if not errors:
for task in tasks:
if 'name' in task:
namelines[task['name']].append(task['__line__'])
if "name" in task:
namelines[task["name"]].append(task["__line__"])
for (name, lines) in namelines.items():
if len(lines) > 1:
errors.append(Error(lines[-1], description % name))
@ -81,28 +81,28 @@ def check_unique_named_task(candidate, settings):
def check_command_instead_of_module(candidate, settings):
tasks, errors = get_normalized_tasks(candidate, settings)
commands = ['command', 'shell', 'raw']
commands = ["command", "shell", "raw"]
modules = {
'git': 'git', 'hg': 'hg', 'curl': 'get_url or uri', 'wget': 'get_url or uri',
'svn': 'subversion', 'service': 'service', 'mount': 'mount',
'rpm': 'yum or rpm_key', 'yum': 'yum', 'apt-get': 'apt-get',
'unzip': 'unarchive', 'tar': 'unarchive', 'chkconfig': 'service',
'rsync': 'synchronize', 'supervisorctl': 'supervisorctl', 'systemctl': 'systemd',
'sed': 'template or lineinfile'
"git": "git", "hg": "hg", "curl": "get_url or uri", "wget": "get_url or uri",
"svn": "subversion", "service": "service", "mount": "mount",
"rpm": "yum or rpm_key", "yum": "yum", "apt-get": "apt-get",
"unzip": "unarchive", "tar": "unarchive", "chkconfig": "service",
"rsync": "synchronize", "supervisorctl": "supervisorctl", "systemctl": "systemd",
"sed": "template or lineinfile"
}
description = "%s command used in place of %s module"
if not errors:
for task in tasks:
if task["action"]["__ansible_module__"] in commands:
if 'cmd' in task['action']:
if "cmd" in task["action"]:
first_cmd_arg = task["action"]["cmd"].split()[0]
else:
first_cmd_arg = task["action"]["__ansible_arguments__"][0]
executable = os.path.basename(first_cmd_arg)
if (first_cmd_arg and executable in modules
and task['action'].get('warn', True) and 'register' not in task):
and task["action"].get("warn", True) and "register" not in task):
errors.append(
Error(task["__line__"], description % (executable, modules[executable])))
@ -111,10 +111,10 @@ def check_command_instead_of_module(candidate, settings):
def check_install_use_latest(candidate, settings):
tasks, errors = get_normalized_tasks(candidate, settings)
package_managers = ['yum', 'apt', 'dnf', 'homebrew', 'pacman', 'openbsd_package', 'pkg5',
'portage', 'pkgutil', 'slackpkg', 'swdepot', 'zypper', 'bundler', 'pip',
'pear', 'npm', 'yarn', 'gem', 'easy_install', 'bower', 'package', 'apk',
'openbsd_pkg', 'pkgng', 'sorcery', 'xbps']
package_managers = ["yum", "apt", "dnf", "homebrew", "pacman", "openbsd_package", "pkg5",
"portage", "pkgutil", "slackpkg", "swdepot", "zypper", "bundler", "pip",
"pear", "npm", "yarn", "gem", "easy_install", "bower", "package", "apk",
"openbsd_pkg", "pkgng", "sorcery", "xbps"]
description = "package installs should use state=present with or without a version"
if not errors:
@ -132,14 +132,14 @@ def check_shell_instead_command(candidate, settings):
if not errors:
for task in tasks:
if task["action"]["__ansible_module__"] == 'shell':
if 'cmd' in task['action']:
if task["action"]["__ansible_module__"] == "shell":
if "cmd" in task["action"]:
cmd = task["action"].get("cmd", [])
else:
cmd = ' '.join(task["action"].get("__ansible_arguments__", []))
cmd = " ".join(task["action"].get("__ansible_arguments__", []))
unjinja = re.sub(r"\{\{[^\}]*\}\}", "JINJA_VAR", cmd)
if not any([ch in unjinja for ch in '&|<>;$\n*[]{}?']):
if not any([ch in unjinja for ch in "&|<>;$\n*[]{}?"]):
errors.append(Error(task["__line__"], description))
return Result(candidate.path, errors)
@ -147,7 +147,7 @@ def check_shell_instead_command(candidate, settings):
def check_command_has_changes(candidate, settings):
tasks, errors = get_normalized_tasks(candidate, settings)
commands = ['command', 'shell', 'raw']
commands = ["command", "shell", "raw"]
description = "commands should either read information (and thus set changed_when) or not " \
"do something if it has already been done (using creates/removes) " \
"or only do it if another check has a particular result (when)"
@ -155,10 +155,10 @@ def check_command_has_changes(candidate, settings):
if not errors:
for task in tasks:
if task["action"]["__ansible_module__"] in commands:
if ('changed_when' not in task and 'when' not in task
and 'when' not in task['__ansible_action_meta__']
and 'creates' not in task['action']
and 'removes' not in task['action']):
if ("changed_when" not in task and "when" not in task
and "when" not in task["__ansible_action_meta__"]
and "creates" not in task["action"]
and "removes" not in task["action"]):
errors.append(Error(task["__line__"], description))
return Result(candidate.path, errors)
@ -166,8 +166,8 @@ def check_command_has_changes(candidate, settings):
def check_empty_string_compare(candidate, settings):
yamllines, errors = get_normalized_yaml(candidate, settings)
description = 'use `when: var` rather than `when: var != ""` (or ' \
'conversely `when: not var` rather than `when: var == ""`)'
description = "use `when: var` rather than `when: var != ""` (or " \
"conversely `when: not var` rather than `when: var == ""`)"
empty_string_compare = re.compile("[=!]= ?[\"'][\"']")
@ -202,7 +202,7 @@ def check_delegate_to_localhost(candidate, settings):
if not errors:
for task in tasks:
if task.get('delegate_to') == 'localhost':
if task.get("delegate_to") == "localhost":
errors.append(Error(task["__line__"], description))
return Result(candidate.path, errors)
@ -225,12 +225,12 @@ def check_literal_bool_format(candidate, settings):
def check_become_user(candidate, settings):
tasks, errors = get_normalized_tasks(candidate, settings)
description = "the task has 'become:' enabled but 'become_user:' is missing"
true_value = [True, 'true', 'True', 'TRUE', 'yes', 'Yes', 'YES']
true_value = [True, "true", "True", "TRUE", "yes", "Yes", "YES"]
if not errors:
gen = (task for task in tasks if 'become' in task)
gen = (task for task in tasks if "become" in task)
for task in gen:
if task["become"] in true_value and 'become_user' not in task.keys():
if task["become"] in true_value and "become_user" not in task.keys():
errors.append(Error(task["__line__"], description))
return Result(candidate.path, errors)

4
ansiblelater/rules/rolefiles.py
View File

@ -23,7 +23,7 @@ def check_scm_in_src(candidate, settings):
if not errors:
for role in roles:
if '+' in role.get('src'):
errors.append(Error(role['__line__'], description))
if "+" in role.get("src"):
errors.append(Error(role["__line__"], description))
return Result(candidate.path, errors)

3
ansiblelater/rules/taskfiles.py
View File

@ -1,5 +1,6 @@
import re
"""Checks related to ansible task files."""
import re
from collections import defaultdict
from ansiblelater.command.review import Error, Result

23
ansiblelater/rules/yamlfiles.py
View File

@ -1,16 +1,15 @@
import codecs
import yaml
import os
from ansiblelater.command.review import Result, Error
from ansiblelater.utils.rulehelper import get_action_tasks
from ansiblelater.utils.rulehelper import get_normalized_yaml
from ansiblelater.utils.rulehelper import get_normalized_task
from ansiblelater.utils.rulehelper import run_yamllint
import yaml
from ansiblelater.command.review import Error, Result
from ansiblelater.utils.rulehelper import (get_action_tasks,
get_normalized_task,
get_normalized_yaml, run_yamllint)
def check_yaml_has_content(candidate, settings):
Testvar = "test"
lines, errors = get_normalized_yaml(candidate, settings)
description = "the file appears to have no useful content"
@ -31,16 +30,16 @@ def check_native_yaml(candidate, settings):
errors.extend(error)
break
action = normal_form['action']['__ansible_module__']
arguments = normal_form['action']['__ansible_arguments__']
# Cope with `set_fact` where task['set_fact'] is None
action = normal_form["action"]["__ansible_module__"]
arguments = normal_form["action"]["__ansible_arguments__"]
# Cope with `set_fact` where task["set_fact"] is None
if not task.get(action):
continue
if isinstance(task[action], dict):
continue
# strip additional newlines off task[action]
if task[action].strip().split() != arguments:
errors.append(Error(task['__line__'], description))
errors.append(Error(task["__line__"], description))
return Result(candidate.path, errors)
@ -82,7 +81,7 @@ def check_yaml_file(candidate, settings):
errors.append(
Error(None, "file does not have a .yml extension"))
elif os.path.isfile(filename) and os.path.splitext(filename)[1][1:] == "yml":
with codecs.open(filename, mode='rb', encoding='utf-8') as f:
with codecs.open(filename, mode="rb", encoding="utf-8") as f:
try:
yaml.safe_load(f)
except Exception as e:

3
ansiblelater/standard.py
View File

@ -1,3 +1,6 @@
"""Standard definition."""
class Standard(object):
"""
Standard definition for all defined rules.

2
ansiblelater/utils/__init__.py
View File

@ -41,7 +41,7 @@ def get_property(prop):
parentdir = os.path.dirname(currentdir)
result = re.search(
r'{}\s*=\s*[\'"]([^\'"]*)[\'"]'.format(prop),
open(os.path.join(parentdir, '__init__.py')).read())
open(os.path.join(parentdir, "__init__.py")).read())
return result.group(1)