363 lines
14 KiB
Python
Executable File
363 lines
14 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
# Copyright 2014 Mirantis, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import argparse
|
|
import ast
|
|
import importlib
|
|
import inspect
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
|
|
from oslo_utils import uuidutils
|
|
import six.moves.urllib.parse as urlparse
|
|
|
|
DECORATOR_MODULE = 'decorators'
|
|
DECORATOR_NAME = 'idempotent_id'
|
|
DECORATOR_IMPORT = 'tempest.%s' % DECORATOR_MODULE
|
|
IMPORT_LINE = 'from tempest.lib import %s' % DECORATOR_MODULE
|
|
DECORATOR_TEMPLATE = "@%s.%s('%%s')" % (DECORATOR_MODULE,
|
|
DECORATOR_NAME)
|
|
UNIT_TESTS_EXCLUDE = 'tempest.tests'
|
|
|
|
|
|
class SourcePatcher(object):
|
|
|
|
""""Lazy patcher for python source files"""
|
|
|
|
def __init__(self):
|
|
self.source_files = None
|
|
self.patches = None
|
|
self.clear()
|
|
|
|
def clear(self):
|
|
"""Clear inner state"""
|
|
self.source_files = {}
|
|
self.patches = {}
|
|
|
|
@staticmethod
|
|
def _quote(s):
|
|
return urlparse.quote(s)
|
|
|
|
@staticmethod
|
|
def _unquote(s):
|
|
return urlparse.unquote(s)
|
|
|
|
def add_patch(self, filename, patch, line_no):
|
|
"""Add lazy patch"""
|
|
if filename not in self.source_files:
|
|
with open(filename) as f:
|
|
self.source_files[filename] = self._quote(f.read())
|
|
patch_id = uuidutils.generate_uuid()
|
|
if not patch.endswith('\n'):
|
|
patch += '\n'
|
|
self.patches[patch_id] = self._quote(patch)
|
|
lines = self.source_files[filename].split(self._quote('\n'))
|
|
lines[line_no - 1] = ''.join(('{%s:s}' % patch_id, lines[line_no - 1]))
|
|
self.source_files[filename] = self._quote('\n').join(lines)
|
|
|
|
@staticmethod
|
|
def _save_changes(filename, source):
|
|
print('%s fixed' % filename)
|
|
with open(filename, 'w') as f:
|
|
f.write(source)
|
|
|
|
def apply_patches(self):
|
|
"""Apply all patches"""
|
|
for filename in self.source_files:
|
|
patched_source = self._unquote(
|
|
self.source_files[filename].format(**self.patches)
|
|
)
|
|
self._save_changes(filename, patched_source)
|
|
self.clear()
|
|
|
|
|
|
class TestChecker(object):
|
|
|
|
def __init__(self, package):
|
|
self.package = package
|
|
self.base_path = os.path.abspath(os.path.dirname(package.__file__))
|
|
|
|
def _path_to_package(self, path):
|
|
relative_path = path[len(self.base_path) + 1:]
|
|
if relative_path:
|
|
return '.'.join((self.package.__name__,) +
|
|
tuple(relative_path.split('/')))
|
|
else:
|
|
return self.package.__name__
|
|
|
|
def _modules_search(self):
|
|
"""Recursive search for python modules in base package"""
|
|
modules = []
|
|
for root, _, files in os.walk(self.base_path):
|
|
if not os.path.exists(os.path.join(root, '__init__.py')):
|
|
continue
|
|
root_package = self._path_to_package(root)
|
|
for item in files:
|
|
if item.endswith('.py'):
|
|
module_name = '.'.join((root_package,
|
|
os.path.splitext(item)[0]))
|
|
if not module_name.startswith(UNIT_TESTS_EXCLUDE):
|
|
modules.append(module_name)
|
|
return modules
|
|
|
|
@staticmethod
|
|
def _get_idempotent_id(test_node):
|
|
"Return key-value dict with metadata from @decorators.idempotent_id"
|
|
idempotent_id = None
|
|
for decorator in test_node.decorator_list:
|
|
if (hasattr(decorator, 'func') and
|
|
hasattr(decorator.func, 'attr') and
|
|
decorator.func.attr == DECORATOR_NAME and
|
|
hasattr(decorator.func, 'value') and
|
|
decorator.func.value.id == DECORATOR_MODULE):
|
|
for arg in decorator.args:
|
|
idempotent_id = ast.literal_eval(arg)
|
|
return idempotent_id
|
|
|
|
@staticmethod
|
|
def _is_decorator(line):
|
|
return line.strip().startswith('@')
|
|
|
|
@staticmethod
|
|
def _is_def(line):
|
|
return line.strip().startswith('def ')
|
|
|
|
def _add_uuid_to_test(self, patcher, test_node, source_path):
|
|
with open(source_path) as src:
|
|
src_lines = src.read().split('\n')
|
|
lineno = test_node.lineno
|
|
insert_position = lineno
|
|
while True:
|
|
if (self._is_def(src_lines[lineno - 1]) or
|
|
(self._is_decorator(src_lines[lineno - 1]) and
|
|
(DECORATOR_TEMPLATE.split('(')[0] <=
|
|
src_lines[lineno - 1].strip().split('(')[0]))):
|
|
insert_position = lineno
|
|
break
|
|
lineno += 1
|
|
patcher.add_patch(
|
|
source_path,
|
|
' ' * test_node.col_offset + DECORATOR_TEMPLATE % uuid.uuid4(),
|
|
insert_position
|
|
)
|
|
|
|
@staticmethod
|
|
def _is_test_case(module, node):
|
|
if (node.__class__ is ast.ClassDef and
|
|
hasattr(module, node.name) and
|
|
inspect.isclass(getattr(module, node.name))):
|
|
return issubclass(getattr(module, node.name), unittest.TestCase)
|
|
|
|
@staticmethod
|
|
def _is_test_method(node):
|
|
return (node.__class__ is ast.FunctionDef and
|
|
node.name.startswith('test_'))
|
|
|
|
@staticmethod
|
|
def _next_node(body, node):
|
|
if body.index(node) < len(body):
|
|
return body[body.index(node) + 1]
|
|
|
|
@staticmethod
|
|
def _import_name(node):
|
|
if isinstance(node, ast.Import):
|
|
return node.names[0].name
|
|
elif isinstance(node, ast.ImportFrom):
|
|
return '%s.%s' % (node.module, node.names[0].name)
|
|
|
|
def _add_import_for_test_uuid(self, patcher, src_parsed, source_path):
|
|
with open(source_path) as f:
|
|
src_lines = f.read().split('\n')
|
|
line_no = 0
|
|
tempest_imports = [node for node in src_parsed.body
|
|
if self._import_name(node) and
|
|
'tempest.' in self._import_name(node)]
|
|
if not tempest_imports:
|
|
import_snippet = '\n'.join(('', IMPORT_LINE, ''))
|
|
else:
|
|
for node in tempest_imports:
|
|
if self._import_name(node) < DECORATOR_IMPORT:
|
|
continue
|
|
else:
|
|
line_no = node.lineno
|
|
import_snippet = IMPORT_LINE
|
|
break
|
|
else:
|
|
line_no = tempest_imports[-1].lineno
|
|
while True:
|
|
if (not src_lines[line_no - 1] or
|
|
getattr(self._next_node(src_parsed.body,
|
|
tempest_imports[-1]),
|
|
'lineno') == line_no or
|
|
line_no == len(src_lines)):
|
|
break
|
|
line_no += 1
|
|
import_snippet = '\n'.join((IMPORT_LINE, ''))
|
|
patcher.add_patch(source_path, import_snippet, line_no)
|
|
|
|
def get_tests(self):
|
|
"""Get test methods with sources from base package with metadata"""
|
|
tests = {}
|
|
for module_name in self._modules_search():
|
|
tests[module_name] = {}
|
|
module = importlib.import_module(module_name)
|
|
source_path = '.'.join(
|
|
(os.path.splitext(module.__file__)[0], 'py')
|
|
)
|
|
with open(source_path, 'r') as f:
|
|
source = f.read()
|
|
tests[module_name]['source_path'] = source_path
|
|
tests[module_name]['tests'] = {}
|
|
source_parsed = ast.parse(source)
|
|
tests[module_name]['ast'] = source_parsed
|
|
tests[module_name]['import_valid'] = (
|
|
hasattr(module, DECORATOR_MODULE) and
|
|
inspect.ismodule(getattr(module, DECORATOR_MODULE))
|
|
)
|
|
test_cases = (node for node in source_parsed.body
|
|
if self._is_test_case(module, node))
|
|
for node in test_cases:
|
|
for subnode in filter(self._is_test_method, node.body):
|
|
test_name = '%s.%s' % (node.name, subnode.name)
|
|
tests[module_name]['tests'][test_name] = subnode
|
|
return tests
|
|
|
|
@staticmethod
|
|
def _filter_tests(function, tests):
|
|
"""Filter tests with condition 'function(test_node) == True'"""
|
|
result = {}
|
|
for module_name in tests:
|
|
for test_name in tests[module_name]['tests']:
|
|
if function(module_name, test_name, tests):
|
|
if module_name not in result:
|
|
result[module_name] = {
|
|
'ast': tests[module_name]['ast'],
|
|
'source_path': tests[module_name]['source_path'],
|
|
'import_valid': tests[module_name]['import_valid'],
|
|
'tests': {}
|
|
}
|
|
result[module_name]['tests'][test_name] = \
|
|
tests[module_name]['tests'][test_name]
|
|
return result
|
|
|
|
def find_untagged(self, tests):
|
|
"""Filter all tests without uuid in metadata"""
|
|
def check_uuid_in_meta(module_name, test_name, tests):
|
|
idempotent_id = self._get_idempotent_id(
|
|
tests[module_name]['tests'][test_name])
|
|
return not idempotent_id
|
|
return self._filter_tests(check_uuid_in_meta, tests)
|
|
|
|
def report_collisions(self, tests):
|
|
"""Reports collisions if there are any
|
|
|
|
Returns true if collisions exist.
|
|
"""
|
|
uuids = {}
|
|
|
|
def report(module_name, test_name, tests):
|
|
test_uuid = self._get_idempotent_id(
|
|
tests[module_name]['tests'][test_name])
|
|
if not test_uuid:
|
|
return
|
|
if test_uuid in uuids:
|
|
error_str = "%s:%s\n uuid %s collision: %s<->%s\n%s:%s" % (
|
|
tests[module_name]['source_path'],
|
|
tests[module_name]['tests'][test_name].lineno,
|
|
test_uuid,
|
|
test_name,
|
|
uuids[test_uuid]['test_name'],
|
|
uuids[test_uuid]['source_path'],
|
|
uuids[test_uuid]['test_node'].lineno,
|
|
)
|
|
print(error_str)
|
|
print("cannot automatically resolve the collision, please "
|
|
"manually remove the duplicate value on the new test.")
|
|
return True
|
|
else:
|
|
uuids[test_uuid] = {
|
|
'module': module_name,
|
|
'test_name': test_name,
|
|
'test_node': tests[module_name]['tests'][test_name],
|
|
'source_path': tests[module_name]['source_path']
|
|
}
|
|
return bool(self._filter_tests(report, tests))
|
|
|
|
def report_untagged(self, tests):
|
|
"""Reports untagged tests if there are any
|
|
|
|
Returns true if untagged tests exist.
|
|
"""
|
|
def report(module_name, test_name, tests):
|
|
error_str = ("%s:%s\nmissing @decorators.idempotent_id"
|
|
"('...')\n%s\n") % (
|
|
tests[module_name]['source_path'],
|
|
tests[module_name]['tests'][test_name].lineno,
|
|
test_name
|
|
)
|
|
print(error_str)
|
|
return True
|
|
return bool(self._filter_tests(report, tests))
|
|
|
|
def fix_tests(self, tests):
|
|
"""Add uuids to all specified in tests and fix it in source files"""
|
|
patcher = SourcePatcher()
|
|
for module_name in tests:
|
|
add_import_once = True
|
|
for test_name in tests[module_name]['tests']:
|
|
if not tests[module_name]['import_valid'] and add_import_once:
|
|
self._add_import_for_test_uuid(
|
|
patcher,
|
|
tests[module_name]['ast'],
|
|
tests[module_name]['source_path']
|
|
)
|
|
add_import_once = False
|
|
self._add_uuid_to_test(
|
|
patcher, tests[module_name]['tests'][test_name],
|
|
tests[module_name]['source_path'])
|
|
patcher.apply_patches()
|
|
|
|
|
|
def run():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--package', action='store', dest='package',
|
|
default='tempest', type=str,
|
|
help='Package with tests')
|
|
parser.add_argument('--fix', action='store_true', dest='fix_tests',
|
|
help='Attempt to fix tests without UUIDs')
|
|
args = parser.parse_args()
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
|
|
pkg = importlib.import_module(args.package)
|
|
checker = TestChecker(pkg)
|
|
errors = False
|
|
tests = checker.get_tests()
|
|
untagged = checker.find_untagged(tests)
|
|
errors = checker.report_collisions(tests) or errors
|
|
if args.fix_tests and untagged:
|
|
checker.fix_tests(untagged)
|
|
else:
|
|
errors = checker.report_untagged(untagged) or errors
|
|
if errors:
|
|
sys.exit("@decorators.idempotent_id existence and uniqueness checks "
|
|
"failed\n"
|
|
"Run 'tox -v -e uuidgen' to automatically fix tests with\n"
|
|
"missing @decorators.idempotent_id decorators.")
|
|
|
|
if __name__ == '__main__':
|
|
run()
|