Skip to content

Commit

Permalink
Fix packaging (six)
Browse files Browse the repository at this point in the history 
This fixes #683 six bug which is only another side-effect of the weird packaging implementation changes made to ansible-lint in its more recent history.

This change aims to address these issues by implementing proven working packaging practices, while removing custom hacks.

We will still need to do some follow-ups to remove implementation code from __init__.py which is a known source of problems. This patch avoided that in order to minimize changes.

Fixes: #683
Signed-off-by: Sorin Sbarnea <ssbarnea@redhat.com>
  • Loading branch information
@ssbarnea
ssbarnea committed Feb 28, 2020
1 parent 5765765 commit a23c89c
Show file tree
Hide file tree
Showing 7 changed files with 293 additions and 309 deletions.
278 changes: 15 additions & 263 deletions lib/ansiblelint/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,274 +17,26 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

from __future__ import print_function
from collections import defaultdict
from __future__ import absolute_import
import os
import re
import sys

import six

import ansiblelint.utils
import codecs

default_rulesdir = os.path.join(os.path.dirname(ansiblelint.utils.__file__), 'rules')


class AnsibleLintRule(object):

def __repr__(self):
return self.id + ": " + self.shortdesc

def verbose(self):
return self.id + ": " + self.shortdesc + "\n " + self.description

match = None
matchtask = None
matchplay = None

@staticmethod
def unjinja(text):
return re.sub(r"{{[^}]*}}", "JINJA_VAR", text)

def matchlines(self, file, text):
matches = []
if not self.match:
return matches
# arrays are 0-based, line numbers are 1-based
# so use prev_line_no as the counter
for (prev_line_no, line) in enumerate(text.split("\n")):
if line.lstrip().startswith('#'):
continue

rule_id_list = ansiblelint.utils.get_rule_skips_from_line(line)
if self.id in rule_id_list:
continue

result = self.match(file, line)
if not result:
continue
message = None
if isinstance(result, six.string_types):
message = result
matches.append(Match(prev_line_no + 1, line,
file['path'], self, message))
return matches

def matchtasks(self, file, text):
matches = []
if not self.matchtask:
return matches

if file['type'] == 'meta':
return matches

yaml = ansiblelint.utils.parse_yaml_linenumbers(text, file['path'])
if not yaml:
return matches

yaml = ansiblelint.utils.append_skipped_rules(yaml, text, file['type'])

for task in ansiblelint.utils.get_normalized_tasks(yaml, file):
if self.id in task.get('skipped_rules', ()):
continue

if 'action' not in task:
continue
result = self.matchtask(file, task)
if not result:
continue

message = None
if isinstance(result, six.string_types):
message = result
task_msg = "Task/Handler: " + ansiblelint.utils.task_to_str(task)
matches.append(Match(task[ansiblelint.utils.LINE_NUMBER_KEY], task_msg,
file['path'], self, message))
return matches

def matchyaml(self, file, text):
matches = []
if not self.matchplay:
return matches

yaml = ansiblelint.utils.parse_yaml_linenumbers(text, file['path'])
if not yaml:
return matches

if isinstance(yaml, dict):
yaml = [yaml]

yaml = ansiblelint.utils.append_skipped_rules(yaml, text, file['type'])

for play in yaml:
if self.id in play.get('skipped_rules', ()):
continue

result = self.matchplay(file, play)
if not result:
continue

if isinstance(result, tuple):
result = [result]

if not isinstance(result, list):
raise TypeError("{} is not a list".format(result))

for section, message in result:
matches.append(Match(play[ansiblelint.utils.LINE_NUMBER_KEY],
section, file['path'], self, message))
return matches


class RulesCollection(object):

def __init__(self, rulesdirs=None):
if rulesdirs is None:
rulesdirs = []
self.rulesdirs = ansiblelint.utils.expand_paths_vars(rulesdirs)
self.rules = []
for rulesdir in self.rulesdirs:
self.extend(ansiblelint.utils.load_plugins(rulesdir))

def register(self, obj):
self.rules.append(obj)

def __iter__(self):
return iter(self.rules)

def __len__(self):
return len(self.rules)

def extend(self, more):
self.rules.extend(more)

def run(self, playbookfile, tags=set(), skip_list=frozenset()):
text = ""
matches = list()

try:
with codecs.open(playbookfile['path'], mode='rb', encoding='utf-8') as f:
text = f.read()
except IOError as e:
print("WARNING: Couldn't open %s - %s" %
(playbookfile['path'], e.strerror),
file=sys.stderr)
return matches

for rule in self.rules:
if not tags or not set(rule.tags).union([rule.id]).isdisjoint(tags):
rule_definition = set(rule.tags)
rule_definition.add(rule.id)
if set(rule_definition).isdisjoint(skip_list):
matches.extend(rule.matchlines(playbookfile, text))
matches.extend(rule.matchtasks(playbookfile, text))
matches.extend(rule.matchyaml(playbookfile, text))

return matches

def __repr__(self):
return "\n".join([rule.verbose()
for rule in sorted(self.rules, key=lambda x: x.id)])

def listtags(self):
tags = defaultdict(list)
for rule in self.rules:
for tag in rule.tags:
tags[tag].append("[{0}]".format(rule.id))
results = []
for tag in sorted(tags):
results.append("{0} {1}".format(tag, tags[tag]))
return "\n".join(results)


class Match(object):

def __init__(self, linenumber, line, filename, rule, message=None):
self.linenumber = linenumber
self.line = line
self.filename = filename
self.rule = rule
self.message = message or rule.shortdesc

def __repr__(self):
formatstr = u"[{0}] ({1}) matched {2}:{3} {4}"
return formatstr.format(self.rule.id, self.message,
self.filename, self.linenumber, self.line)


class Runner(object):

def __init__(self, rules, playbook, tags, skip_list, exclude_paths,
verbosity=0, checked_files=None):
self.rules = rules
self.playbooks = set()
# assume role if directory
if os.path.isdir(playbook):
self.playbooks.add((os.path.join(playbook, ''), 'role'))
self.playbook_dir = playbook
else:
self.playbooks.add((playbook, 'playbook'))
self.playbook_dir = os.path.dirname(playbook)
self.tags = tags
self.skip_list = skip_list
self._update_exclude_paths(exclude_paths)
self.verbosity = verbosity
if checked_files is None:
checked_files = set()
self.checked_files = checked_files

def _update_exclude_paths(self, exclude_paths):
if exclude_paths:
# These will be (potentially) relative paths
paths = ansiblelint.utils.expand_paths_vars(exclude_paths)
# Since ansiblelint.utils.find_children returns absolute paths,
# and the list of files we create in `Runner.run` can contain both
# relative and absolute paths, we need to cover both bases.
self.exclude_paths = paths + [os.path.abspath(p) for p in paths]
else:
self.exclude_paths = []
__metaclass__ = type

def is_excluded(self, file_path):
# Any will short-circuit as soon as something returns True, but will
# be poor performance for the case where the path under question is
# not excluded.
return any(file_path.startswith(path) for path in self.exclude_paths)
try:
import pkg_resources

def run(self):
files = list()
for playbook in self.playbooks:
if self.is_excluded(playbook[0]):
continue
if playbook[1] == 'role':
continue
files.append({'path': ansiblelint.utils.normpath(playbook[0]), 'type': playbook[1]})
visited = set()
while (visited != self.playbooks):
for arg in self.playbooks - visited:
for child in ansiblelint.utils.find_children(arg, self.playbook_dir):
if self.is_excluded(child['path']):
continue
self.playbooks.add((child['path'], child['type']))
files.append(child)
visited.add(arg)
__version__ = pkg_resources.get_distribution("molecule").version
except Exception:
__version__ = "unknown"

matches = list()

# remove duplicates from files list
files = [value for n, value in enumerate(files) if value not in files[:n]]
# this file needs to be importable even if dependencies are not present
# in order to avoid packging errors in some cases.
try:
import ansiblelint.utils
from ansiblelint.base import AnsibleLintRule, RulesCollection, Match, Runner # noqa

# remove files that have already been checked
files = [x for x in files if x['path'] not in self.checked_files]
for file in files:
if self.verbosity > 0:
print("Examining %s of type %s" % (
ansiblelint.utils.normpath(file['path']),
file['type']))
matches.extend(self.rules.run(file, tags=set(self.tags),
skip_list=self.skip_list))
# update list of checked files
self.checked_files.update([x['path'] for x in files])
default_rulesdir = os.path.join(os.path.dirname(ansiblelint.utils.__file__), 'rules')

return matches
except ImportError:
pass

0 comments on commit a23c89c

Please sign in to comment.