ansible-later/ansiblelater/command/candidates.py

391 lines
12 KiB
Python
Raw Normal View History

2019-04-05 20:05:06 +00:00
"""Review candidates."""
2019-04-02 14:34:03 +00:00
import codecs
2019-04-03 15:42:46 +00:00
import copy
2019-04-02 14:34:03 +00:00
import os
import re
from distutils.version import LooseVersion
2019-04-05 20:05:06 +00:00
from six import iteritems
from ansiblelater import LOG
from ansiblelater import utils
from ansiblelater.logger import flag_extra
2019-04-02 14:34:03 +00:00
try:
# Ansible 2.4 import of module loader
from ansible.plugins.loader import module_loader
except ImportError:
try:
from ansible.plugins import module_loader
except ImportError:
from ansible.utils import module_finder as module_loader
class Candidate(object):
"""
Meta object for all files which later has to process.
Each file passed to later will be classified by type and
bundled with necessary meta informations for rule processing.
"""
def __init__(self, filename, settings={}, standards=[]):
self.path = filename
self.binary = False
self.vault = False
self.filetype = type(self).__name__.lower()
self.expected_version = True
2019-04-03 15:42:46 +00:00
self.standards = self._get_standards(settings, standards)
2019-04-02 14:34:03 +00:00
try:
with codecs.open(filename, mode="rb", encoding="utf-8") as f:
if f.readline().startswith("$ANSIBLE_VAULT"):
self.vault = True
except UnicodeDecodeError:
self.binary = True
self.version = self._get_version(settings)
2019-04-02 14:34:03 +00:00
def _get_version(self, settings):
path = self.path
2019-04-02 14:34:03 +00:00
version = None
if not self.binary:
if isinstance(self, RoleFile):
parentdir = os.path.dirname(os.path.abspath(self.path))
while parentdir != os.path.dirname(parentdir):
meta_file = os.path.join(parentdir, "meta", "main.yml")
if os.path.exists(meta_file):
path = meta_file
break
parentdir = os.path.dirname(parentdir)
version_re = re.compile(r"^# Standards:\s*([\d.]+)")
with codecs.open(path, mode="rb", encoding="utf-8") as f:
for line in f:
match = version_re.match(line)
if match:
version = match.group(1)
2019-04-02 14:34:03 +00:00
if not version:
2019-04-03 15:42:46 +00:00
version = utils.standards_latest(self.standards)
2019-04-02 14:34:03 +00:00
if self.expected_version:
if isinstance(self, RoleFile):
2019-04-11 13:56:20 +00:00
LOG.warning(
"%s %s is in a role that contains a meta/main.yml without a declared "
"standards version. "
"Using latest standards version %s" %
2020-04-05 12:33:43 +00:00
(type(self).__name__, self.path, version)
)
2019-04-02 14:34:03 +00:00
else:
2019-04-11 13:56:20 +00:00
LOG.warning(
"%s %s does not present standards version. "
"Using latest standards version %s" %
2020-04-05 12:33:43 +00:00
(type(self).__name__, self.path, version)
)
2019-04-05 12:02:14 +00:00
else:
2020-04-05 12:33:43 +00:00
LOG.info(
"%s %s declares standards version %s" % (type(self).__name__, self.path, version)
)
2019-04-02 14:34:03 +00:00
return version
2019-04-03 15:42:46 +00:00
def _get_standards(self, settings, standards):
target_standards = []
2019-04-17 10:33:23 +00:00
includes = settings.config["rules"]["filter"]
excludes = settings.config["rules"]["exclude_filter"]
2019-04-03 15:42:46 +00:00
2019-04-17 10:33:23 +00:00
if len(includes) == 0:
includes = [s.id for s in standards]
for standard in standards:
if standard.id in includes and standard.id not in excludes:
target_standards.append(standard)
2019-04-03 15:42:46 +00:00
return target_standards
2019-04-16 09:33:21 +00:00
def review(self, settings, lines=None):
2019-04-02 14:34:03 +00:00
errors = 0
2019-04-03 15:42:46 +00:00
for standard in self.standards:
if type(self).__name__.lower() not in standard.types:
2019-04-02 14:34:03 +00:00
continue
2019-04-15 15:26:02 +00:00
2019-04-03 15:42:46 +00:00
result = standard.check(self, settings.config)
2019-04-02 14:34:03 +00:00
if not result:
2020-04-05 12:33:43 +00:00
utils.sysexit_with_message(
"Standard '{}' returns an empty result object.".format(
standard.check.__name__
)
)
labels = {
"tag": "review",
"standard": standard.name,
"file": self.path,
"passed": True
}
2019-04-03 15:42:46 +00:00
if standard.id and standard.id.strip():
labels["id"] = standard.id
2020-04-05 12:33:43 +00:00
for err in [
err for err in result.errors if not err.lineno
or utils.is_line_in_ranges(err.lineno, utils.lines_ranges(lines))
]: # noqa
2019-04-03 15:42:46 +00:00
err_labels = copy.copy(labels)
err_labels["passed"] = False
if isinstance(err, Error):
err_labels.update(err.to_dict())
2019-04-02 14:34:03 +00:00
if not standard.version:
2020-04-05 12:33:43 +00:00
LOG.warning(
"{id}Best practice '{name}' not met:\n{path}:{error}".format(
id=self._format_id(standard.id),
name=standard.name,
path=self.path,
error=err
),
extra=flag_extra(err_labels)
)
2019-04-03 15:42:46 +00:00
elif LooseVersion(standard.version) > LooseVersion(self.version):
2020-04-05 12:33:43 +00:00
LOG.warning(
"{id}Future standard '{name}' not met:\n{path}:{error}".format(
id=self._format_id(standard.id),
name=standard.name,
path=self.path,
error=err
),
extra=flag_extra(err_labels)
)
2019-04-02 14:34:03 +00:00
else:
2020-04-05 12:33:43 +00:00
LOG.error(
"{id}Standard '{name}' not met:\n{path}:{error}".format(
id=self._format_id(standard.id),
name=standard.name,
path=self.path,
error=err
),
extra=flag_extra(err_labels)
)
2019-04-02 14:34:03 +00:00
errors = errors + 1
2019-04-15 15:26:02 +00:00
return errors
def _format_id(self, standard_id):
if standard_id and standard_id.strip():
standard_id = "[{id}] ".format(id=standard_id.strip())
return standard_id
2020-04-05 12:33:43 +00:00
def __repr__(self): # noqa
2019-04-02 14:34:03 +00:00
return "%s (%s)" % (type(self).__name__, self.path)
2020-04-05 12:33:43 +00:00
def __getitem__(self, item): # noqa
2019-04-02 14:34:03 +00:00
return self.__dict__.get(item)
class RoleFile(Candidate):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible role file."""
2020-04-05 12:33:43 +00:00
2019-04-02 14:34:03 +00:00
def __init__(self, filename, settings={}, standards=[]):
super(RoleFile, self).__init__(filename, settings, standards)
2019-04-03 15:42:46 +00:00
parentdir = os.path.dirname(os.path.abspath(filename))
while parentdir != os.path.dirname(parentdir):
role_modules = os.path.join(parentdir, "library")
if os.path.exists(role_modules):
module_loader.add_directory(role_modules)
break
parentdir = os.path.dirname(parentdir)
2019-04-02 14:34:03 +00:00
class Playbook(Candidate):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible playbook."""
2019-04-02 14:34:03 +00:00
pass
class Task(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible task file."""
2020-04-05 12:33:43 +00:00
2019-04-02 14:34:03 +00:00
def __init__(self, filename, settings={}, standards=[]):
super(Task, self).__init__(filename, settings, standards)
self.filetype = "tasks"
class Handler(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible handler file."""
2020-04-05 12:33:43 +00:00
2019-04-02 14:34:03 +00:00
def __init__(self, filename, settings={}, standards=[]):
super(Handler, self).__init__(filename, settings, standards)
self.filetype = "handlers"
class Vars(Candidate):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible vars file."""
2019-04-02 14:34:03 +00:00
pass
class Unversioned(Candidate):
2020-04-05 12:54:39 +00:00
"""Object classified as unversioned file."""
2020-04-05 12:33:43 +00:00
2019-04-02 14:34:03 +00:00
def __init__(self, filename, settings={}, standards=[]):
super(Unversioned, self).__init__(filename, settings, standards)
self.expected_version = False
class InventoryVars(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible inventory vars."""
2019-04-02 14:34:03 +00:00
pass
class HostVars(InventoryVars):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible host vars."""
2019-04-02 14:34:03 +00:00
pass
class GroupVars(InventoryVars):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible group vars."""
2019-04-02 14:34:03 +00:00
pass
class RoleVars(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible role vars."""
2019-04-02 14:34:03 +00:00
pass
class Meta(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible meta file."""
2019-04-02 14:34:03 +00:00
pass
class Inventory(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible inventory file."""
2019-04-02 14:34:03 +00:00
pass
class Code(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as code file."""
2019-04-02 14:34:03 +00:00
pass
class Template(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible template file."""
2019-04-02 14:34:03 +00:00
pass
class Doc(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as documentation file."""
2019-04-02 14:34:03 +00:00
pass
class Makefile(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as makefile."""
2019-04-02 14:34:03 +00:00
pass
class File(RoleFile):
2020-04-05 12:54:39 +00:00
"""Object classified as generic file."""
2019-04-02 14:34:03 +00:00
pass
class Rolesfile(Unversioned):
2020-04-05 12:54:39 +00:00
"""Object classified as Ansible roles file."""
2019-04-02 14:34:03 +00:00
pass
2019-04-05 20:05:06 +00:00
class Error(object):
"""Default error object created if a rule failed."""
def __init__(self, lineno, message, error_type=None, **kwargs):
"""
Initialize a new error object and returns None.
:param lineno: Line number where the error from de rule occures
:param message: Detailed error description provided by the rule
"""
self.lineno = lineno
self.message = message
self.kwargs = kwargs
for (key, value) in iteritems(kwargs):
setattr(self, key, value)
2020-04-05 12:33:43 +00:00
def __repr__(self): # noqa
2019-04-05 20:05:06 +00:00
if self.lineno:
return "%s: %s" % (self.lineno, self.message)
else:
return " %s" % (self.message)
def to_dict(self):
result = dict(lineno=self.lineno, message=self.message)
for (key, value) in iteritems(self.kwargs):
result[key] = value
return result
class Result(object):
2020-04-05 12:54:39 +00:00
"""Generic result object."""
2020-04-05 12:33:43 +00:00
2019-04-05 20:05:06 +00:00
def __init__(self, candidate, errors=None):
self.candidate = candidate
self.errors = errors or []
def message(self):
2020-04-05 12:33:43 +00:00
return "\n".join(["{0}:{1}".format(self.candidate, error) for error in self.errors])
2019-04-05 20:05:06 +00:00
2019-04-02 14:34:03 +00:00
def classify(filename, settings={}, standards=[]):
parentdir = os.path.basename(os.path.dirname(filename))
basename = os.path.basename(filename)
if parentdir in ["tasks"]:
return Task(filename, settings, standards)
if parentdir in ["handlers"]:
return Handler(filename, settings, standards)
if parentdir in ["vars", "defaults"]:
return RoleVars(filename, settings, standards)
if "group_vars" in filename.split(os.sep):
return GroupVars(filename, settings, standards)
if "host_vars" in filename.split(os.sep):
return HostVars(filename, settings, standards)
if parentdir in ["meta"]:
return Meta(filename, settings, standards)
2020-04-05 12:54:39 +00:00
if (
parentdir in ["library", "lookup_plugins", "callback_plugins", "filter_plugins"]
or filename.endswith(".py")
):
2019-04-02 14:34:03 +00:00
return Code(filename, settings, standards)
if "inventory" == basename or "hosts" == basename or parentdir in ["inventories"]:
2019-04-02 14:34:03 +00:00
return Inventory(filename, settings, standards)
if "rolesfile" in basename or "requirements" in basename:
return Rolesfile(filename, settings, standards)
if "Makefile" in basename:
return Makefile(filename, settings, standards)
if "templates" in filename.split(os.sep) or basename.endswith(".j2"):
return Template(filename, settings, standards)
if "files" in filename.split(os.sep):
return File(filename, settings, standards)
if basename.endswith(".yml") or basename.endswith(".yaml"):
return Playbook(filename, settings, standards)
if "README" in basename:
return Doc(filename, settings, standards)
return None