Add hacking checks

Added required hacking checks to follow OpenStack coding
standards.

Change-Id: I20fe168f364296844a593ec3880e1639637dbf06
This commit is contained in:
Abhishek Kekane 2016-07-13 18:11:15 +05:30
parent d4f055262e
commit 48051ece73
8 changed files with 1093 additions and 3 deletions

View File

@ -1,4 +1,50 @@
masakari Style Commandments
===============================================
===========================
Read the OpenStack Style Commandments http://docs.openstack.org/developer/hacking/
- Step 1: Read the OpenStack Style Commandments
http://docs.openstack.org/developer/hacking/
- Step 2: Read on
Masakari Specific Commandments
------------------------------
- [M301] no db session in public API methods (disabled)
This enforces a guideline defined in ``oslo.db.sqlalchemy.session``
- [M302] timeutils.utcnow() wrapper must be used instead of direct
calls to datetime.datetime.utcnow() to make it easy to override its return value in tests
- [M303] capitalize help string
Config parameter help strings should have a capitalized first letter
- [M304] vim configuration should not be kept in source files.
- [M305] Change assertTrue(isinstance(A, B)) by optimal assert like
assertIsInstance(A, B).
- [M306] Change assertEqual(type(A), B) by optimal assert like
assertIsInstance(A, B)
- [M307] Change assertEqual(A, None) or assertEqual(None, A) by optimal assert like
assertIsNone(A)
- [M308] Validate that debug level logs are not translated.
- [M309] Don't import translation in tests
- [M310] Setting CONF.* attributes directly in tests is forbidden. Use
self.flags(option=value) instead.
- [M311] Validate that LOG.info messages use _LI.
- [M312] Validate that LOG.exception messages use _LE.
- [M313] Validate that LOG.warning and LOG.warn messages use _LW.
- [M314] Log messages require translations!
- [M315] Method's default argument shouldn't be mutable
- [M316] Ensure that the _() function is explicitly imported to ensure proper translations.
- [M317] Ensure that jsonutils.%(fun)s must be used instead of json.%(fun)s
- [M318] Change assertTrue/False(A in/not in B, message) to the more specific
assertIn/NotIn(A, B, message)
- [M319] Check for usage of deprecated assertRaisesRegexp
- [M320] Must use a dict comprehension instead of a dict constructor with a sequence of key-value pairs.
- [M321] Change assertEqual(A in B, True), assertEqual(True, A in B),
assertEqual(A in B, False) or assertEqual(False, A in B) to the more specific
assertIn/NotIn(A, B)
- [M322] Check masakari.utils.spawn() is used instead of greenthread.spawn() and eventlet.spawn()
- [M323] contextlib.nested is deprecated
- [M324] Config options should be in the central location ``masakari/conf/``
- [M325] Check for common double word typos
- [M326] Python 3: do not use dict.iteritems.
- [M327] Python 3: do not use dict.iterkeys.
- [M328] Python 3: do not use dict.itervalues.
- [M329] Deprecated library function os.popen()

View File

@ -18,6 +18,7 @@ from oslo_log import log as logging
import six.moves.urllib.parse as urlparse
import masakari.conf
from masakari.i18n import _
CONF = masakari.conf.CONF

View File

447
masakari/hacking/checks.py Normal file
View File

@ -0,0 +1,447 @@
# Copyright (c) 2016, NTT Data
# All Rights Reserved.
#
# 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 re
import pep8
"""
Guidelines for writing new hacking checks
- Use only for Masakari specific tests. OpenStack general tests
should be submitted to the common 'hacking' module.
- Pick numbers in the range M3xx. Find the current test with
the highest allocated number and then pick the next value.
- Keep the test method code in the source file ordered based
on the M3xx value.
- List the new rule in the top level HACKING.rst file
- Add test cases for each new rule to masakari/tests/unit/test_hacking.py
"""
UNDERSCORE_IMPORT_FILES = []
session_check = re.compile(r"\w*def [a-zA-Z0-9].*[(].*session.*[)]")
cfg_re = re.compile(r".*\scfg\.")
cfg_opt_re = re.compile(r".*[\s\[]cfg\.[a-zA-Z]*Opt\(")
vi_header_re = re.compile(r"^#\s+vim?:.+")
asse_trueinst_re = re.compile(
r"(.)*assertTrue\(isinstance\((\w|\.|\'|\"|\[|\])+, "
"(\w|\.|\'|\"|\[|\])+\)\)")
asse_equal_type_re = re.compile(
r"(.)*assertEqual\(type\((\w|\.|\'|\"|\[|\])+\), "
"(\w|\.|\'|\"|\[|\])+\)")
asse_equal_in_end_with_true_or_false_re = re.compile(
r"assertEqual\("r"(\w|[][.'\"])+ in (\w|[][.'\", ])+, (True|False)\)")
asse_equal_in_start_with_true_or_false_re = re.compile(
r"assertEqual\("r"(True|False), (\w|[][.'\"])+ in (\w|[][.'\", ])+\)")
asse_equal_end_with_none_re = re.compile(
r"assertEqual\(.*?,\s+None\)$")
asse_equal_start_with_none_re = re.compile(
r"assertEqual\(None,")
# NOTE(abhishekk): Next two regexes weren't united to one for more readability.
# asse_true_false_with_in_or_not_in regex checks
# assertTrue/False(A in B) cases where B argument has no spaces
# asse_true_false_with_in_or_not_in_spaces regex checks cases
# where B argument has spaces and starts/ends with [, ', ".
# For example: [1, 2, 3], "some string", 'another string'.
# We have to separate these regexes to escape a false positives
# results. B argument should have spaces only if it starts
# with [, ", '. Otherwise checking of string
# "assertFalse(A in B and C in D)" will be false positives.
# In this case B argument is "B and C in D".
asse_true_false_with_in_or_not_in = re.compile(
r"assert(True|False)\("r"(\w|[][.'\"])+( not)? in (\w|[][.'\",])"
r"+(, .*)?\)")
asse_true_false_with_in_or_not_in_spaces = re.compile(
r"assert(True|False)"r"\((\w|[][.'\"])+( not)? in [\[|'|\"](\w|"
r"[][.'\", ])+[\[|'|\"](, .*)?\)")
asse_raises_regexp = re.compile(r"assertRaisesRegexp\(")
conf_attribute_set_re = re.compile(r"CONF\.[a-z0-9_.]+\s*=\s*\w")
log_translation = re.compile(
r"(.)*LOG\.(audit|error|critical)\(\s*('|\")")
log_translation_info = re.compile(
r"(.)*LOG\.(info)\(\s*(_\(|'|\")")
log_translation_exception = re.compile(
r"(.)*LOG\.(exception)\(\s*(_\(|'|\")")
log_translation_LW = re.compile(
r"(.)*LOG\.(warning|warn)\(\s*(_\(|'|\")")
translated_log = re.compile(
r"(.)*LOG\.(audit|error|info|critical|exception)"
"\(\s*_\(\s*('|\")")
mutable_default_args = re.compile(r"^\s*def .+\((.+=\{\}|.+=\[\])")
string_translation = re.compile(r"[^_]*_\(\s*('|\")")
underscore_import_check = re.compile(r"(.)*import _(.)*")
import_translation_for_log_or_exception = re.compile(
r"(.)*(from\smasakari.i18n\simport)\s_")
# We need this for cases where they have created their own _ function.
custom_underscore_check = re.compile(r"(.)*_\s*=\s*(.)*")
dict_constructor_with_list_copy_re = re.compile(r".*\bdict\((\[)?(\(|\[)")
http_not_implemented_re = re.compile(r"raise .*HTTPNotImplemented\(")
spawn_re = re.compile(
r".*(eventlet|greenthread)\.(?P<spawn_part>spawn(_n)?)\(.*\)")
contextlib_nested = re.compile(r"^with (contextlib\.)?nested\(")
doubled_words_re = re.compile(
r"\b(then?|[iao]n|i[fst]|but|f?or|at|and|[dt]o)\s+\1\b")
def no_db_session_in_public_api(logical_line, filename):
if "db/api.py" in filename:
if session_check.match(logical_line):
yield (0, "M301: public db api methods may not accept"
" session")
def use_timeutils_utcnow(logical_line, filename):
# tools are OK to use the standard datetime module
if "/tools/" in filename:
return
msg = ("M302: timeutils.utcnow() must be used instead of "
"datetime.%s()")
datetime_funcs = ['now', 'utcnow']
for f in datetime_funcs:
pos = logical_line.find('datetime.%s' % f)
if pos != -1:
yield (pos, msg % f)
def capital_cfg_help(logical_line, tokens):
msg = "M303: capitalize help string"
if cfg_re.match(logical_line):
for t in range(len(tokens)):
if tokens[t][1] == "help":
txt = tokens[t + 2][1]
if len(txt) > 1 and txt[1].islower():
yield(0, msg)
def no_vi_headers(physical_line, line_number, lines):
"""Check for vi editor configuration in source files.
By default vi modelines can only appear in the first or
last 5 lines of a source file.
M304
"""
# NOTE(abhishekk): line_number is 1-indexed
if line_number <= 5 or line_number > len(lines) - 5:
if vi_header_re.match(physical_line):
return 0, "M304: Don't put vi configuration in source files"
def assert_true_instance(logical_line):
"""Check for assertTrue(isinstance(a, b)) sentences
M305
"""
if asse_trueinst_re.match(logical_line):
yield (0, "M305: assertTrue(isinstance(a, b)) sentences "
"not allowed")
def assert_equal_type(logical_line):
"""Check for assertEqual(type(A), B) sentences
M306
"""
if asse_equal_type_re.match(logical_line):
yield (0, "M306: assertEqual(type(A), B) sentences not allowed")
def assert_equal_none(logical_line):
"""Check for assertEqual(A, None) or assertEqual(None, A) sentences
M307
"""
res = (asse_equal_start_with_none_re.search(logical_line) or
asse_equal_end_with_none_re.search(logical_line))
if res:
yield (0, "M307: assertEqual(A, None) or assertEqual(None, A) "
"sentences not allowed")
def no_translate_debug_logs(logical_line, filename):
"""Check for 'LOG.debug(_('
As per our translation policy,
https://wiki.openstack.org/wiki/LoggingStandards#Log_Translation
we shouldn't translate debug level logs.
* This check assumes that 'LOG' is a logger.
* Use filename so we can start enforcing this in specific folders instead
of needing to do so all at once.
M308
"""
if logical_line.startswith("LOG.debug(_("):
yield(0, "M308 Don't translate debug level logs")
def no_import_translation_in_tests(logical_line, filename):
"""Check for 'from masakari.i18n import _'
M309
"""
if 'masakari/tests/' in filename:
res = import_translation_for_log_or_exception.match(logical_line)
if res:
yield(0, "M309 Don't import translation in tests")
def no_setting_conf_directly_in_tests(logical_line, filename):
"""Check for setting CONF.* attributes directly in tests
The value can leak out of tests affecting how subsequent tests run.
Using self.flags(option=value) is the preferred method to temporarily
set config options in tests.
M310
"""
if 'masakari/tests/' in filename:
res = conf_attribute_set_re.match(logical_line)
if res:
yield (0, "M310: Setting CONF.* attributes directly in "
"tests is forbidden. Use self.flags(option=value) "
"instead")
def validate_log_translations(logical_line, physical_line, filename):
# Translations are not required in the test directory
if "masakari/tests" in filename:
return
if pep8.noqa(physical_line):
return
msg = "M311: LOG.info messages require translations `_LI()`!"
if log_translation_info.match(logical_line):
yield (0, msg)
msg = "M312: LOG.exception messages require translations `_LE()`!"
if log_translation_exception.match(logical_line):
yield (0, msg)
msg = ("M313: LOG.warning, LOG.warn messages require "
"translations `_LW()`!")
if log_translation_LW.match(logical_line):
yield (0, msg)
msg = "M314: Log messages require translations!"
if log_translation.match(logical_line):
yield (0, msg)
def no_mutable_default_args(logical_line):
msg = "M315: Method's default argument shouldn't be mutable!"
if mutable_default_args.match(logical_line):
yield (0, msg)
def check_explicit_underscore_import(logical_line, filename):
"""Check for explicit import of the _ function
We need to ensure that any files that are using the _() function
to translate logs are explicitly importing the _ function. We
can't trust unit test to catch whether the import has been
added so we need to check for it here.
"""
# Build a list of the files that have _ imported. No further
# checking needed once it is found.
if filename in UNDERSCORE_IMPORT_FILES:
pass
elif (underscore_import_check.match(logical_line) or
custom_underscore_check.match(logical_line)):
UNDERSCORE_IMPORT_FILES.append(filename)
elif (translated_log.match(logical_line) or
string_translation.match(logical_line)):
yield(0, "M316: Found use of _() without explicit "
"import of _ !")
def use_jsonutils(logical_line, filename):
# tools are OK to use the standard json module
if "/tools/" in filename:
return
msg = "M317: jsonutils.%(fun)s must be used instead of json.%(fun)s"
if "json." in logical_line:
json_funcs = ['dumps(', 'dump(', 'loads(', 'load(']
for f in json_funcs:
pos = logical_line.find('json.%s' % f)
if pos != -1:
yield (pos, msg % {'fun': f[:-1]})
def assert_true_or_false_with_in(logical_line):
"""Check for assertTrue/False(A in B), assertTrue/False(A not in B),
assertTrue/False(A in B, message) or assertTrue/False(A not in B, message)
sentences.
M318
"""
res = (asse_true_false_with_in_or_not_in.search(logical_line) or
asse_true_false_with_in_or_not_in_spaces.search(logical_line))
if res:
yield (0, "M318: Use assertIn/NotIn(A, B) rather than "
"assertTrue/False(A in/not in B) when checking collection "
"contents.")
def assert_raises_regexp(logical_line):
"""Check for usage of deprecated assertRaisesRegexp
M319
"""
res = asse_raises_regexp.search(logical_line)
if res:
yield (0, "M319: assertRaisesRegex must be used instead "
"of assertRaisesRegexp")
def dict_constructor_with_list_copy(logical_line):
msg = ("M320: Must use a dict comprehension instead of a dict "
"constructor with a sequence of key-value pairs.")
if dict_constructor_with_list_copy_re.match(logical_line):
yield (0, msg)
def assert_equal_in(logical_line):
"""Check for assertEqual(A in B, True), assertEqual(True, A in B),
assertEqual(A in B, False) or assertEqual(False, A in B) sentences
M321
"""
res = (asse_equal_in_start_with_true_or_false_re.search(logical_line) or
asse_equal_in_end_with_true_or_false_re.search(logical_line))
if res:
yield (0, "M321: Use assertIn/NotIn(A, B) rather than "
"assertEqual(A in B, True/False) when checking collection "
"contents.")
def check_greenthread_spawns(logical_line, physical_line, filename):
"""Check for use of greenthread.spawn(), greenthread.spawn_n(),
eventlet.spawn(), and eventlet.spawn_n()
M322
"""
msg = ("M322: Use masakari.utils.%(spawn)s() rather than "
"greenthread.%(spawn)s() and eventlet.%(spawn)s()")
if "masakari/utils.py" in filename or "masakari/tests/" in filename:
return
match = re.match(spawn_re, logical_line)
if match:
yield (0, msg % {'spawn': match.group('spawn_part')})
def check_no_contextlib_nested(logical_line, filename):
msg = ("M323: contextlib.nested is deprecated. With Python 2.7"
"and later the with-statement supports multiple nested objects. "
"See https://docs.python.org/2/library/contextlib.html"
"#contextlib.nested for more information. masakari.test.nested() "
"is an alternative as well.")
if contextlib_nested.match(logical_line):
yield(0, msg)
def check_config_option_in_central_place(logical_line, filename):
msg = ("M324: Config options should be in the central location "
"'/masakari/conf/*'. Do not declare new config options outside "
"of that folder.")
# That's the correct location
if "masakari/conf/" in filename:
return
if cfg_opt_re.match(logical_line):
yield(0, msg)
def check_doubled_words(physical_line, filename):
"""Check for the common doubled-word typos
M325
"""
msg = ("M325: Doubled word '%(word)s' typo found")
match = re.search(doubled_words_re, physical_line)
if match:
return (0, msg % {'word': match.group(1)})
def check_python3_no_iteritems(logical_line):
msg = ("M326: Use six.iteritems() instead of dict.iteritems().")
if re.search(r".*\.iteritems\(\)", logical_line):
yield(0, msg)
def check_python3_no_iterkeys(logical_line):
msg = ("M327: Use six.iterkeys() instead of dict.iterkeys().")
if re.search(r".*\.iterkeys\(\)", logical_line):
yield(0, msg)
def check_python3_no_itervalues(logical_line):
msg = ("M328: Use six.itervalues() instead of dict.itervalues().")
if re.search(r".*\.itervalues\(\)", logical_line):
yield(0, msg)
def no_os_popen(logical_line):
"""Disallow 'os.popen('
Deprecated library function os.popen() Replace it using subprocess
https://bugs.launchpad.net/tempest/+bug/1529836
M329
"""
if 'os.popen(' in logical_line:
yield(0, 'M329 Deprecated library function os.popen(). '
'Replace it using subprocess module. ')
def factory(register):
register(no_db_session_in_public_api)
register(use_timeutils_utcnow)
register(capital_cfg_help)
register(no_vi_headers)
register(no_import_translation_in_tests)
register(assert_true_instance)
register(assert_equal_type)
register(assert_equal_none)
register(assert_raises_regexp)
register(no_translate_debug_logs)
register(no_setting_conf_directly_in_tests)
register(validate_log_translations)
register(no_mutable_default_args)
register(check_explicit_underscore_import)
register(use_jsonutils)
register(assert_true_or_false_with_in)
register(dict_constructor_with_list_copy)
register(assert_equal_in)
register(check_no_contextlib_nested)
register(check_greenthread_spawns)
register(check_config_option_in_central_place)
register(check_doubled_words)
register(check_python3_no_iteritems)
register(check_python3_no_iterkeys)
register(check_python3_no_itervalues)
register(no_os_popen)

82
masakari/test.py Normal file
View File

@ -0,0 +1,82 @@
# Copyright 2016 NTT Data
# All Rights Reserved.
#
# 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.
"""Base classes for our unit tests.
Allows overriding of flags for use of fakes, and some black magic for
inline callbacks.
"""
import contextlib
import eventlet
eventlet.monkey_patch(os=False)
import mock
import six
import testtools
if six.PY2:
nested = contextlib.nested
else:
@contextlib.contextmanager
def nested(*contexts):
with contextlib.ExitStack() as stack:
yield [stack.enter_context(c) for c in contexts]
def _patch_mock_to_raise_for_invalid_assert_calls():
def raise_for_invalid_assert_calls(wrapped):
def wrapper(_self, name):
valid_asserts = [
'assert_called_with',
'assert_called_once_with',
'assert_has_calls',
'assert_any_calls']
if name.startswith('assert') and name not in valid_asserts:
raise AttributeError('%s is not a valid mock assert method'
% name)
return wrapped(_self, name)
return wrapper
mock.Mock.__getattr__ = raise_for_invalid_assert_calls(
mock.Mock.__getattr__)
# NOTE(abhishekk): needs to be called only once at import time
# to patch the mock lib
_patch_mock_to_raise_for_invalid_assert_calls()
class TestCase(testtools.TestCase):
"""Test case base class for all unit tests.
Due to the slowness of DB access, please consider deriving from
`NoDBTestCase` first.
"""
USES_DB = True
def setUp(self):
"""Run before each test method to initialize test environment."""
super(TestCase, self).setUp()
class NoDBTestCase(TestCase):
"""`NoDBTestCase` differs from TestCase in that DB access is not supported.
This makes tests run significantly faster. If possible, all new tests
should derive from this class.
"""
USES_DB = False

View File

@ -0,0 +1,26 @@
# Copyright 2016 NTT Data
# All Rights Reserved.
#
# 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.
"""
:mod:`masakari.tests.unit` -- Masakari Unittests
=====================================================
.. automodule:: nova.tests.unit
:platform: Unix
"""
import eventlet
eventlet.monkey_patch(os=False)

View File

@ -0,0 +1,483 @@
# Copyright 2016 NTT Data.
#
# 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 textwrap
import mock
import pep8
from masakari.hacking import checks
from masakari import test
class HackingTestCase(test.NoDBTestCase):
"""This class tests the hacking checks in masakari.hacking.checks by
passing strings to the check methods like the pep8/flake8 parser would.
The parser loops over each line in the file and then passes the
parameters to the check method. The parameter names in the check method
dictate what type of object is passed to the check method.
The parameter types are::
logical_line: A processed line with the following modifications:
- Multi-line statements converted to a single line.
- Stripped left and right.
- Contents of strings replaced with "xxx" of same length.
- Comments removed.
physical_line: Raw line of text from the input file.
lines: a list of the raw lines from the input file
tokens: the tokens that contribute to this logical line
line_number: line number in the input file
total_lines: number of lines in the input file
blank_lines: blank lines before this one
indent_char: indentation character in this file (" " or "\t")
indent_level: indentation (with tabs expanded to multiples of 8)
previous_indent_level: indentation on previous line
previous_logical: previous logical line
filename: Path of the file being run through pep8
When running a test on a check method the return will be False/None if
there is no violation in the sample input. If there is an error a tuple is
returned with a position in the line, and a message. So to check the result
just assertTrue if the check is expected to fail and assertFalse if it
should pass.
"""
def test_no_vi_headers(self):
lines = ['Line 1\n', 'Line 2\n', 'Line 3\n', 'Line 4\n', 'Line 5\n',
'Line 6\n', 'Line 7\n', 'Line 8\n', 'Line 9\n', 'Line 10\n',
'Line 11\n', 'Line 12\n', 'Line 13\n', 'Line14\n', 'Line15\n']
self.assertIsNone(checks.no_vi_headers(
"Test string foo", 1, lines))
self.assertEqual(len(list(checks.no_vi_headers(
"# vim: et tabstop=4 shiftwidth=4 softtabstop=4",
2, lines))), 2)
self.assertIsNone(checks.no_vi_headers(
"# vim: et tabstop=4 shiftwidth=4 softtabstop=4",
6, lines))
self.assertIsNone(checks.no_vi_headers(
"# vim: et tabstop=4 shiftwidth=4 softtabstop=4",
9, lines))
self.assertEqual(len(list(checks.no_vi_headers(
"# vim: et tabstop=4 shiftwidth=4 softtabstop=4",
14, lines))), 2)
self.assertIsNone(checks.no_vi_headers(
"Test end string for vi",
15, lines))
def test_assert_true_instance(self):
self.assertEqual(len(list(checks.assert_true_instance(
"self.assertTrue(isinstance(e, "
"exception.BuildAbortException))"))), 1)
self.assertEqual(
len(list(checks.assert_true_instance("self.assertTrue()"))), 0)
def test_assert_equal_type(self):
self.assertEqual(len(list(checks.assert_equal_type(
"self.assertEqual(type(als['QuicAssist']), list)"))), 1)
self.assertEqual(
len(list(checks.assert_equal_type("self.assertTrue()"))), 0)
def test_assert_equal_in(self):
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(a in b, True)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual('str' in 'string', True)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(any(a==1 for a in b), True)"))), 0)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(True, a in b)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(True, 'str' in 'string')"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(True, any(a==1 for a in b))"))), 0)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(a in b, False)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual('str' in 'string', False)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(any(a==1 for a in b), False)"))), 0)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(False, a in b)"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(False, 'str' in 'string')"))), 1)
self.assertEqual(len(list(checks.assert_equal_in(
"self.assertEqual(False, any(a==1 for a in b))"))), 0)
def test_assert_equal_none(self):
self.assertEqual(len(list(checks.assert_equal_none(
"self.assertEqual(A, None)"))), 1)
self.assertEqual(len(list(checks.assert_equal_none(
"self.assertEqual(None, A)"))), 1)
self.assertEqual(
len(list(checks.assert_equal_none("self.assertIsNone()"))), 0)
def test_assert_true_or_false_with_in_or_not_in(self):
self.assertEqual(len(list(checks.assert_equal_none(
"self.assertEqual(A, None)"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in B)"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertFalse(A in B)"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A not in B)"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertFalse(A not in B)"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in B, 'some message')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertFalse(A in B, 'some message')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A not in B, 'some message')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertFalse(A not in B, 'some message')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in 'some string with spaces')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in 'some string with spaces')"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in ['1', '2', '3'])"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(A in [1, 2, 3])"))), 1)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(any(A > 5 for A in B))"))), 0)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertTrue(any(A > 5 for A in B), 'some message')"))), 0)
self.assertEqual(len(list(checks.assert_true_or_false_with_in(
"self.assertFalse(some in list1 and some2 in list2)"))), 0)
def test_no_translate_debug_logs(self):
self.assertEqual(len(list(checks.no_translate_debug_logs(
"LOG.debug(_('foo'))", "masakari/foo.py"))), 1)
self.assertEqual(len(list(checks.no_translate_debug_logs(
"LOG.debug('foo')", "masakari/foo.py"))), 0)
self.assertEqual(len(list(checks.no_translate_debug_logs(
"LOG.info(_('foo'))", "masakari/foo.py"))), 0)
def test_no_setting_conf_directly_in_tests(self):
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.option = 1", "masakari/tests/test_foo.py"))), 1)
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.group.option = 1", "masakari/tests/test_foo.py"))), 1)
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.option = foo = 1", "masakari/tests/test_foo.py"))), 1)
# Shouldn't fail with comparisons
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.option == 'foo'", "masakari/tests/test_foo.py"))), 0)
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.option != 1", "masakari/tests/test_foo.py"))), 0)
# Shouldn't fail since not in masakari/tests/
self.assertEqual(len(list(checks.no_setting_conf_directly_in_tests(
"CONF.option = 1", "masakari/compute/foo.py"))), 0)
def test_log_translations(self):
logs = ['audit', 'error', 'info', 'warning', 'critical', 'warn',
'exception']
levels = ['_LI', '_LW', '_LE', '_LC']
debug = "LOG.debug('OK')"
self.assertEqual(
0, len(list(checks.validate_log_translations(debug, debug, 'f'))))
for log in logs:
bad = 'LOG.%s("Bad")' % log
self.assertEqual(1,
len(list(
checks.validate_log_translations(bad, bad, 'f'))))
ok = "LOG.%s('OK') # noqa" % log
self.assertEqual(0,
len(list(
checks.validate_log_translations(ok, ok, 'f'))))
ok = "LOG.%s(variable)" % log
self.assertEqual(0,
len(list(
checks.validate_log_translations(ok, ok, 'f'))))
for level in levels:
ok = "LOG.%s(%s('OK'))" % (log, level)
self.assertEqual(0,
len(list(
checks.validate_log_translations(ok, ok, 'f'))))
def test_no_mutable_default_args(self):
self.assertEqual(1, len(list(checks.no_mutable_default_args(
"def get_info_from_bdm(virt_type, bdm, mapping=[])"))))
self.assertEqual(0, len(list(checks.no_mutable_default_args(
"defined = []"))))
self.assertEqual(0, len(list(checks.no_mutable_default_args(
"defined, undefined = [], {}"))))
def test_check_explicit_underscore_import(self):
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"LOG.info(_('My info message'))",
"masakari/tests/other_files.py"))), 1)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"msg = _('My message')",
"masakari/tests/other_files.py"))), 1)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"from masakari.i18n import _",
"masakari/tests/other_files.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"LOG.info(_('My info message'))",
"masakari/tests/other_files.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"msg = _('My message')",
"masakari/tests/other_files.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"from cinder.i18n import _, _LW",
"masakari/tests/other_files2.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"msg = _('My message')",
"masakari/tests/other_files2.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"_ = translations.ugettext",
"masakari/tests/other_files3.py"))), 0)
self.assertEqual(len(list(checks.check_explicit_underscore_import(
"msg = _('My message')",
"masakari/tests/other_files3.py"))), 0)
# We are patching pep8 so that only the check under test is actually
# installed.
@mock.patch('pep8._checks',
{'physical_line': {}, 'logical_line': {}, 'tree': {}})
def _run_check(self, code, checker, filename=None):
pep8.register_check(checker)
lines = textwrap.dedent(code).strip().splitlines(True)
checker = pep8.Checker(filename=filename, lines=lines)
checker.check_all()
checker.report._deferred_print.sort()
return checker.report._deferred_print
def _assert_has_errors(self, code, checker, expected_errors=None,
filename=None):
actual_errors = [e[:3] for e in
self._run_check(code, checker, filename)]
self.assertEqual(expected_errors or [], actual_errors)
def _assert_has_no_errors(self, code, checker, filename=None):
self._assert_has_errors(code, checker, filename=filename)
def test_oslo_assert_raises_regexp(self):
code = """
self.assertRaisesRegexp(ValueError,
"invalid literal for.*XYZ'$",
int,
'XYZ')
"""
self._assert_has_errors(code, checks.assert_raises_regexp,
expected_errors=[(1, 0, "M319")])
def test_dict_constructor_with_list_copy(self):
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" dict([(i, connect_info[i])"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" attrs = dict([(k, _from_json(v))"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" type_names = dict((value, key) for key, value in"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" dict((value, key) for key, value in"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
"foo(param=dict((k, v) for k, v in bar.items()))"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" dict([[i,i] for i in range(3)])"))))
self.assertEqual(1, len(list(checks.dict_constructor_with_list_copy(
" dd = dict([i,i] for i in range(3))"))))
self.assertEqual(0, len(list(checks.dict_constructor_with_list_copy(
" create_kwargs = dict(snapshot=snapshot,"))))
self.assertEqual(0, len(list(checks.dict_constructor_with_list_copy(
" self._render_dict(xml, data_el, data.__dict__)"))))
def test_check_contextlib_use(self):
code = """
with test.nested(
mock.patch.object(network_model.NetworkInfo, 'hydrate'),
mock.patch.object(objects.InstanceInfoCache, 'save'),
) as (
hydrate_mock, save_mock
)
"""
filename = "masakari/api/openstack/ha/test.py"
self._assert_has_no_errors(code, checks.check_no_contextlib_nested,
filename=filename)
code = """
with contextlib.nested(
mock.patch.object(network_model.NetworkInfo, 'hydrate'),
mock.patch.object(objects.InstanceInfoCache, 'save'),
) as (
hydrate_mock, save_mock
)
"""
filename = "masakari/api/openstack/compute/ha/test.py"
errors = [(1, 0, 'M323')]
self._assert_has_errors(code, checks.check_no_contextlib_nested,
expected_errors=errors, filename=filename)
def test_check_greenthread_spawns(self):
errors = [(1, 0, "M322")]
code = "greenthread.spawn(func, arg1, kwarg1=kwarg1)"
self._assert_has_errors(code, checks.check_greenthread_spawns,
expected_errors=errors)
code = "greenthread.spawn_n(func, arg1, kwarg1=kwarg1)"
self._assert_has_errors(code, checks.check_greenthread_spawns,
expected_errors=errors)
code = "eventlet.greenthread.spawn(func, arg1, kwarg1=kwarg1)"
self._assert_has_errors(code, checks.check_greenthread_spawns,
expected_errors=errors)
code = "eventlet.spawn(func, arg1, kwarg1=kwarg1)"
self._assert_has_errors(code, checks.check_greenthread_spawns,
expected_errors=errors)
code = "eventlet.spawn_n(func, arg1, kwarg1=kwarg1)"
self._assert_has_errors(code, checks.check_greenthread_spawns,
expected_errors=errors)
code = "masakari.utils.spawn(func, arg1, kwarg1=kwarg1)"
self._assert_has_no_errors(code, checks.check_greenthread_spawns)
code = "masakari.utils.spawn_n(func, arg1, kwarg1=kwarg1)"
self._assert_has_no_errors(code, checks.check_greenthread_spawns)
def test_config_option_regex_match(self):
def should_match(code):
self.assertTrue(checks.cfg_opt_re.match(code))
def should_not_match(code):
self.assertFalse(checks.cfg_opt_re.match(code))
should_match("opt = cfg.StrOpt('opt_name')")
should_match("opt = cfg.IntOpt('opt_name')")
should_match("opt = cfg.DictOpt('opt_name')")
should_match("opt = cfg.Opt('opt_name')")
should_match("opts=[cfg.Opt('opt_name')]")
should_match(" cfg.Opt('opt_name')")
should_not_match("opt_group = cfg.OptGroup('opt_group_name')")
def test_check_config_option_in_central_place(self):
errors = [(1, 0, "M324")]
code = """
opts = [
cfg.StrOpt('random_opt',
default='foo',
help='I am here to do stuff'),
]
"""
# option at the right place in the tree
self._assert_has_no_errors(code,
checks.check_config_option_in_central_place,
filename="masakari/conf/serial_console.py")
self._assert_has_errors(code,
checks.check_config_option_in_central_place,
filename="masakari/cmd/serialproxy.py",
expected_errors=errors)
def test_check_doubled_words(self):
errors = [(1, 0, "M325")]
# Artificial break to stop pep8 detecting the test !
code = "This is the" + " the best comment"
self._assert_has_errors(code, checks.check_doubled_words,
expected_errors=errors)
code = "This is the then best comment"
self._assert_has_no_errors(code, checks.check_doubled_words)
def test_dict_iteritems(self):
self.assertEqual(1, len(list(checks.check_python3_no_iteritems(
"obj.iteritems()"))))
self.assertEqual(0, len(list(checks.check_python3_no_iteritems(
"six.iteritems(ob))"))))
def test_dict_iterkeys(self):
self.assertEqual(1, len(list(checks.check_python3_no_iterkeys(
"obj.iterkeys()"))))
self.assertEqual(0, len(list(checks.check_python3_no_iterkeys(
"six.iterkeys(ob))"))))
def test_dict_itervalues(self):
self.assertEqual(1, len(list(checks.check_python3_no_itervalues(
"obj.itervalues()"))))
self.assertEqual(0, len(list(checks.check_python3_no_itervalues(
"six.itervalues(ob))"))))
def test_no_os_popen(self):
code = """
import os
foobar_cmd = "foobar -get -beer"
answer = os.popen(foobar_cmd).read()
if answer == nok":
try:
os.popen(os.popen('foobar -beer -please')).read()
except ValueError:
go_home()
"""
errors = [(4, 0, 'M329'), (8, 8, 'M329')]
self._assert_has_errors(code, checks.no_os_popen,
expected_errors=errors)

View File

@ -18,6 +18,7 @@ install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstac
[testenv:pep8]
commands = flake8 {posargs}
deps = hacking
[testenv:pep8-constraints]
install_command = {[testenv:common-constraints]install_command}
@ -59,6 +60,10 @@ commands = oslo_debug_helper {posargs}
# E123, E125 skipped as they are invalid PEP-8.
show-source = True
ignore = E123,E125,H405
ignore = E123,E125,E128,H405
builtins = _
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build
[hacking]
local-check-factory = masakari.hacking.checks.factory
import_exceptions = masakari.i18n