valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-08 09:23:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into expand-list_pkg-attr-parameter

Browse Source
This commit is contained in:
Silvio Moioli 2017-09-04 15:36:44 +02:00 committed by GitHub
commit 130c1b1c21
28 changed files with 1298 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
salt/cloud/__init__.py
View File

@ -5,7 +5,7 @@ correct cloud modules
'''
# Import python libs
from __future__ import absolute_import, print_function, generators
from __future__ import absolute_import, print_function, generators, unicode_literals
import os
import copy
import glob

4
salt/grains/napalm.py
View File

@ -447,8 +447,8 @@ def optional_args(proxy=None):
device2:
True
'''
opt_args = _get_device_grain('optional_args', proxy=proxy)
if _FORBIDDEN_OPT_ARGS:
opt_args = _get_device_grain('optional_args', proxy=proxy) or {}
if opt_args and _FORBIDDEN_OPT_ARGS:
for arg in _FORBIDDEN_OPT_ARGS:
opt_args.pop(arg, None)
return {'optional_args': opt_args}

2
salt/modules/augeas_cfg.py
View File

@ -200,7 +200,7 @@ def execute(context=None, lens=None, commands=(), load_path=None):
method = METHOD_MAP[cmd]
nargs = arg_map[method]
parts = salt.utils.args.shlex_split(arg, posix=False)
parts = salt.utils.args.shlex_split(arg)
if len(parts) not in nargs:
err = '{0} takes {1} args: {2}'.format(method, nargs, parts)

32
salt/modules/genesis.py
View File

@ -24,6 +24,8 @@ import salt.utils.kickstart
import salt.syspaths
from salt.exceptions import SaltInvocationError
# Import 3rd-party libs
from salt.ext import six
log = logging.getLogger(__name__)
@ -325,6 +327,8 @@ def _bootstrap_yum(
'''
if pkgs is None:
pkgs = []
elif isinstance(pkgs, six.string_types):
pkgs = pkgs.split(',')
default_pkgs = ('yum', 'centos-release', 'iputils')
for pkg in default_pkgs:
@ -333,6 +337,8 @@ def _bootstrap_yum(
if exclude_pkgs is None:
exclude_pkgs = []
elif isinstance(exclude_pkgs, six.string_types):
exclude_pkgs = exclude_pkgs.split(',')
for pkg in exclude_pkgs:
pkgs.remove(pkg)
@ -393,15 +399,27 @@ def _bootstrap_deb(
if repo_url is None:
repo_url = 'http://ftp.debian.org/debian/'
if not salt.utils.which('debootstrap'):
log.error('Required tool debootstrap is not installed.')
return False
if isinstance(pkgs, (list, tuple)):
pkgs = ','.join(pkgs)
if isinstance(exclude_pkgs, (list, tuple)):
exclude_pkgs = ','.join(exclude_pkgs)
deb_args = [
'debootstrap',
'--foreign',
'--arch',
_cmd_quote(arch),
'--include',
] + pkgs + [
'--exclude',
] + exclude_pkgs + [
_cmd_quote(arch)]
if pkgs:
deb_args += ['--include', _cmd_quote(pkgs)]
if exclude_pkgs:
deb_args += ['--exclude', _cmd_quote(exclude_pkgs)]
deb_args += [
_cmd_quote(flavor),
_cmd_quote(root),
_cmd_quote(repo_url),
@ -469,6 +487,8 @@ def _bootstrap_pacman(
if pkgs is None:
pkgs = []
elif isinstance(pkgs, six.string_types):
pkgs = pkgs.split(',')
default_pkgs = ('pacman', 'linux', 'systemd-sysvcompat', 'grub')
for pkg in default_pkgs:
@ -477,6 +497,8 @@ def _bootstrap_pacman(
if exclude_pkgs is None:
exclude_pkgs = []
elif isinstance(exclude_pkgs, six.string_types):
exclude_pkgs = exclude_pkgs.split(',')
for pkg in exclude_pkgs:
pkgs.remove(pkg)

3
salt/modules/inspectlib/collector.py
View File

@ -31,6 +31,7 @@ from salt.modules.inspectlib.entities import (AllowedDir, IgnoredDir, Package,
import salt.utils # Can be removed when reinit_crypto is moved
import salt.utils.files
import salt.utils.fsutils
import salt.utils.path
import salt.utils.stringutils
from salt.exceptions import CommandExecutionError
@ -312,7 +313,7 @@ class Inspector(EnvLoader):
continue
if not valid or not os.path.exists(obj) or not os.access(obj, os.R_OK):
continue
if os.path.islink(obj):
if salt.utils.path.islink(obj):
links.append(obj)
elif os.path.isdir(obj):
dirs.append(obj)

7
salt/modules/inspectlib/kiwiproc.py
View File

@ -17,11 +17,14 @@
# Import python libs
from __future__ import absolute_import
import os
import grp
import pwd
from xml.dom import minidom
import platform
import socket
try:
import grp
import pwd
except ImportError:
pass
# Import salt libs
import salt.utils.files

602
salt/modules/saltcheck.py Normal file
View File

@ -0,0 +1,602 @@
# -*- coding: utf-8 -*-
'''
A module for testing the logic of states and highstates
Saltcheck provides unittest like functionality requiring only the knowledge of salt module execution and yaml.
In order to run state and highstate saltcheck tests a sub-folder of a state must be creaed and named "saltcheck-tests".
Tests for a state should be created in files ending in *.tst and placed in the saltcheck-tests folder.
Multiple tests can be created in a file.
Multiple *.tst files can be created in the saltcheck-tests folder.
The "id" of a test works in the same manner as in salt state files.
They should be unique and descriptive.
Example file system layout:
/srv/salt/apache/
init.sls
config.sls
saltcheck-tests/
pkg_and_mods.tst
config.tst
Saltcheck Test Syntax:
Unique-ID:
module_and_function:
args:
kwargs:
assertion:
expected-return:
Example test 1:
echo-test-hello:
module_and_function: test.echo
args:
- "hello"
kwargs:
assertion: assertEqual
expected-return: 'hello'
:codeauthor: William Cannon <william.cannon@gmail.com>
:maturity: new
'''
from __future__ import absolute_import
import logging
import os
import time
import yaml
try:
import salt.utils
import salt.client
import salt.exceptions
except ImportError:
pass
log = logging.getLogger(__name__)
__virtualname__ = 'saltcheck'
def __virtual__():
'''
Check dependencies - may be useful in future
'''
return __virtualname__
def update_master_cache():
'''
Updates the master cache onto the minion - transfers all salt-check-tests
Should be done one time before running tests, and if tests are updated
Can be automated by setting "auto_update_master_cache: True" in minion config
CLI Example:
salt '*' saltcheck.update_master_cache
'''
__salt__['cp.cache_master']()
return True
def run_test(**kwargs):
'''
Execute one saltcheck test and return result
:param keyword arg test:
CLI Example::
salt '*' saltcheck.run_test
test='{"module_and_function": "test.echo",
"assertion": "assertEqual",
"expected-return": "This works!",
"args":["This works!"] }'
'''
# salt converts the string to a dictionary auto-magically
scheck = SaltCheck()
test = kwargs.get('test', None)
if test and isinstance(test, dict):
return scheck.run_test(test)
else:
return "Test must be a dictionary"
def run_state_tests(state):
'''
Execute all tests for a salt state and return results
Nested states will also be tested
:param str state: the name of a user defined state
CLI Example::
salt '*' saltcheck.run_state_tests postfix
'''
scheck = SaltCheck()
paths = scheck.get_state_search_path_list()
stl = StateTestLoader(search_paths=paths)
results = {}
sls_list = _get_state_sls(state)
for state_name in sls_list:
mypath = stl.convert_sls_to_path(state_name)
stl.add_test_files_for_sls(mypath)
stl.load_test_suite()
results_dict = {}
for key, value in stl.test_dict.items():
result = scheck.run_test(value)
results_dict[key] = result
results[state_name] = results_dict
passed = 0
failed = 0
missing_tests = 0
for state in results:
if len(results[state].items()) == 0:
missing_tests = missing_tests + 1
else:
for dummy, val in results[state].items():
log.info("dummy={}, val={}".format(dummy, val))
if val.startswith('Pass'):
passed = passed + 1
if val.startswith('Fail'):
failed = failed + 1
out_list = []
for key, value in results.items():
out_list.append({key: value})
out_list.sort()
out_list.append({"TEST RESULTS": {'Passed': passed, 'Failed': failed, 'Missing Tests': missing_tests}})
return out_list
def run_highstate_tests():
'''
Execute all tests for a salt highstate and return results
CLI Example::
salt '*' saltcheck.run_highstate_tests
'''
scheck = SaltCheck()
paths = scheck.get_state_search_path_list()
stl = StateTestLoader(search_paths=paths)
results = {}
sls_list = _get_top_states()
all_states = []
for top_state in sls_list:
sls_list = _get_state_sls(top_state)
for state in sls_list:
if state not in all_states:
all_states.append(state)
for state_name in all_states:
mypath = stl.convert_sls_to_path(state_name)
stl.add_test_files_for_sls(mypath)
stl.load_test_suite()
results_dict = {}
for key, value in stl.test_dict.items():
result = scheck.run_test(value)
results_dict[key] = result
results[state_name] = results_dict
passed = 0
failed = 0
missing_tests = 0
for state in results:
if len(results[state].items()) == 0:
missing_tests = missing_tests + 1
else:
for dummy, val in results[state].items():
log.info("dummy={}, val={}".format(dummy, val))
if val.startswith('Pass'):
passed = passed + 1
if val.startswith('Fail'):
failed = failed + 1
out_list = []
for key, value in results.items():
out_list.append({key: value})
out_list.sort()
out_list.append({"TEST RESULTS": {'Passed': passed, 'Failed': failed, 'Missing Tests': missing_tests}})
return out_list
def _is_valid_module(module):
'''return a list of all modules available on minion'''
modules = __salt__['sys.list_modules']()
return bool(module in modules)
def _get_auto_update_cache_value():
'''return the config value of auto_update_master_cache'''
__salt__['config.get']('auto_update_master_cache')
return True
def _is_valid_function(module_name, function):
'''Determine if a function is valid for a module'''
try:
functions = __salt__['sys.list_functions'](module_name)
except salt.exceptions.SaltException:
functions = ["unable to look up functions"]
return "{0}.{1}".format(module_name, function) in functions
def _get_top_states():
''' equivalent to a salt cli: salt web state.show_top'''
alt_states = []
try:
returned = __salt__['state.show_top']()
for i in returned['base']:
alt_states.append(i)
except Exception:
raise
# log.info("top states: {}".format(alt_states))
return alt_states
def _get_state_sls(state):
''' equivalent to a salt cli: salt web state.show_low_sls STATE'''
sls_list_state = []
try:
returned = __salt__['state.show_low_sls'](state)
for i in returned:
if i['__sls__'] not in sls_list_state:
sls_list_state.append(i['__sls__'])
except Exception:
raise
return sls_list_state
class SaltCheck(object):
'''
This class implements the saltcheck
'''
def __init__(self):
# self.sls_list_top = []
self.sls_list_state = []
self.modules = []
self.results_dict = {}
self.results_dict_summary = {}
self.assertions_list = '''assertEqual assertNotEqual
assertTrue assertFalse
assertIn assertNotIn
assertGreater
assertGreaterEqual
assertLess assertLessEqual'''.split()
self.auto_update_master_cache = _get_auto_update_cache_value
# self.salt_lc = salt.client.Caller(mopts=__opts__)
self.salt_lc = salt.client.Caller()
if self.auto_update_master_cache:
update_master_cache()
def __is_valid_test(self, test_dict):
'''Determine if a test contains:
a test name,
a valid module and function,
a valid assertion,
an expected return value'''
tots = 0 # need total of >= 6 to be a valid test
m_and_f = test_dict.get('module_and_function', None)
assertion = test_dict.get('assertion', None)
expected_return = test_dict.get('expected-return', None)
log.info("__is_valid_test has test: {}".format(test_dict))
if m_and_f:
tots += 1
module, function = m_and_f.split('.')
if _is_valid_module(module):
tots += 1
if _is_valid_function(module, function):
tots += 1
log.info("__is_valid_test has valid m_and_f")
if assertion:
tots += 1
if assertion in self.assertions_list:
tots += 1
log.info("__is_valid_test has valid_assertion")
if expected_return:
tots += 1
log.info("__is_valid_test has valid_expected_return")
log.info("__is_valid_test score: {}".format(tots))
return tots >= 6
def call_salt_command(self,
fun,
args,
kwargs):
'''Generic call of salt Caller command'''
value = False
try:
if args and kwargs:
value = self.salt_lc.cmd(fun, *args, **kwargs)
elif args and not kwargs:
value = self.salt_lc.cmd(fun, *args)
elif not args and kwargs:
value = self.salt_lc.cmd(fun, **kwargs)
else:
value = self.salt_lc.cmd(fun)
except salt.exceptions.SaltException:
raise
except Exception:
raise
return value
def run_test(self, test_dict):
'''Run a single saltcheck test'''
if self.__is_valid_test(test_dict):
mod_and_func = test_dict['module_and_function']
args = test_dict.get('args', None)
kwargs = test_dict.get('kwargs', None)
assertion = test_dict['assertion']
expected_return = test_dict['expected-return']
actual_return = self.call_salt_command(mod_and_func, args, kwargs)
if assertion != "assertIn":
expected_return = self.cast_expected_to_returned_type(expected_return, actual_return)
if assertion == "assertEqual":
value = self.__assert_equal(expected_return, actual_return)
elif assertion == "assertNotEqual":
value = self.__assert_not_equal(expected_return, actual_return)
elif assertion == "assertTrue":
value = self.__assert_true(expected_return)
elif assertion == "assertFalse":
value = self.__assert_false(expected_return)
elif assertion == "assertIn":
value = self.__assert_in(expected_return, actual_return)
elif assertion == "assertNotIn":
value = self.__assert_not_in(expected_return, actual_return)
elif assertion == "assertGreater":
value = self.__assert_greater(expected_return, actual_return)
elif assertion == "assertGreaterEqual":
value = self.__assert_greater_equal(expected_return, actual_return)
elif assertion == "assertLess":
value = self.__assert_less(expected_return, actual_return)
elif assertion == "assertLessEqual":
value = self.__assert_less_equal(expected_return, actual_return)
else:
value = "Fail - bas assertion"
else:
return "Fail - invalid test"
return value
@staticmethod
def cast_expected_to_returned_type(expected, returned):
'''
Determine the type of variable returned
Cast the expected to the type of variable returned
'''
ret_type = type(returned)
new_expected = expected
if expected == "False" and ret_type == bool:
expected = False
try:
new_expected = ret_type(expected)
except ValueError:
log.info("Unable to cast expected into type of returned")
log.info("returned = {}".format(returned))
log.info("type of returned = {}".format(type(returned)))
log.info("expected = {}".format(expected))
log.info("type of expected = {}".format(type(expected)))
return new_expected
@staticmethod
def __assert_equal(expected, returned):
'''
Test if two objects are equal
'''
result = "Pass"
try:
assert (expected == returned), "{0} is not equal to {1}".format(expected, returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_not_equal(expected, returned):
'''
Test if two objects are not equal
'''
result = "Pass"
try:
assert (expected != returned), "{0} is equal to {1}".format(expected, returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_true(returned):
'''
Test if an boolean is True
'''
result = "Pass"
try:
assert (returned is True), "{0} not True".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_false(returned):
'''
Test if an boolean is False
'''
result = "Pass"
if isinstance(returned, str):
try:
returned = bool(returned)
except ValueError:
raise
try:
assert (returned is False), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_in(expected, returned):
'''
Test if a value is in the list of returned values
'''
result = "Pass"
try:
assert (expected in returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_not_in(expected, returned):
'''
Test if a value is not in the list of returned values
'''
result = "Pass"
try:
assert (expected not in returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_greater(expected, returned):
'''
Test if a value is greater than the returned value
'''
result = "Pass"
try:
assert (expected > returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_greater_equal(expected, returned):
'''
Test if a value is greater than or equal to the returned value
'''
result = "Pass"
try:
assert (expected >= returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_less(expected, returned):
'''
Test if a value is less than the returned value
'''
result = "Pass"
try:
assert (expected < returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def __assert_less_equal(expected, returned):
'''
Test if a value is less than or equal to the returned value
'''
result = "Pass"
try:
assert (expected <= returned), "{0} not False".format(returned)
except AssertionError as err:
result = "Fail: " + str(err)
return result
@staticmethod
def get_state_search_path_list():
'''For the state file system, return a
list of paths to search for states'''
# state cache should be updated before running this method
search_list = []
cachedir = __opts__.get('cachedir', None)
environment = __opts__['environment']
if environment:
path = cachedir + os.sep + "files" + os.sep + environment
search_list.append(path)
path = cachedir + os.sep + "files" + os.sep + "base"
search_list.append(path)
return search_list
class StateTestLoader(object):
'''
Class loads in test files for a state
e.g. state_dir/saltcheck-tests/[1.tst, 2.tst, 3.tst]
'''
def __init__(self, search_paths):
self.search_paths = search_paths
self.path_type = None
self.test_files = [] # list of file paths
self.test_dict = {}
def load_test_suite(self):
'''load tests either from one file, or a set of files'''
self.test_dict = {}
for myfile in self.test_files:
self.load_file(myfile)
self.test_files = []
def load_file(self, filepath):
'''
loads in one test file
'''
try:
with salt.utils.files.fopen(filepath, 'r') as myfile:
# with open(filepath, 'r') as myfile:
contents_yaml = yaml.load(myfile)
for key, value in contents_yaml.items():
self.test_dict[key] = value
except:
raise
return
def gather_files(self, filepath):
'''gather files for a test suite'''
self.test_files = []
log.info("gather_files: {}".format(time.time()))
filepath = filepath + os.sep + 'saltcheck-tests'
rootdir = filepath
# for dirname, subdirlist, filelist in os.walk(rootdir):
for dirname, dummy, filelist in os.walk(rootdir):
for fname in filelist:
if fname.endswith('.tst'):
start_path = dirname + os.sep + fname
full_path = os.path.abspath(start_path)
self.test_files.append(full_path)
return
@staticmethod
def convert_sls_to_paths(sls_list):
'''Converting sls to paths'''
new_sls_list = []
for sls in sls_list:
sls = sls.replace(".", os.sep)
new_sls_list.append(sls)
return new_sls_list
@staticmethod
def convert_sls_to_path(sls):
'''Converting sls to paths'''
sls = sls.replace(".", os.sep)
return sls
def add_test_files_for_sls(self, sls_path):
'''Adding test files'''
for path in self.search_paths:
full_path = path + os.sep + sls_path
rootdir = full_path
if os.path.isdir(full_path):
log.info("searching path= {}".format(full_path))
# for dirname, subdirlist, filelist in os.walk(rootdir, topdown=True):
for dirname, subdirlist, dummy in os.walk(rootdir, topdown=True):
if "saltcheck-tests" in subdirlist:
self.gather_files(dirname)
log.info("test_files list: {}".format(self.test_files))
log.info("found subdir match in = {}".format(dirname))
else:
log.info("did not find subdir match in = {}".format(dirname))
del subdirlist[:]
else:
log.info("path is not a directory= {}".format(full_path))
return

11
salt/modules/win_pkg.py
View File

@ -977,7 +977,7 @@ def install(name=None, refresh=False, pkgs=None, **kwargs):
# Version is ignored
salt '*' pkg.install pkgs="['foo', 'bar']" version=1.2.3
If passed with a comma seperated list in the ``name`` parameter, the
If passed with a comma separated list in the ``name`` parameter, the
version will apply to all packages in the list.
CLI Example:
@ -1286,7 +1286,7 @@ def install(name=None, refresh=False, pkgs=None, **kwargs):
use_msiexec, msiexec = _get_msiexec(pkginfo[version_num].get('msiexec', False))
# Build cmd and arguments
# cmd and arguments must be seperated for use with the task scheduler
# cmd and arguments must be separated for use with the task scheduler
if use_msiexec:
cmd = msiexec
arguments = ['/i', cached_pkg]
@ -1328,7 +1328,9 @@ def install(name=None, refresh=False, pkgs=None, **kwargs):
# Run Scheduled Task
# Special handling for installing salt
if pkg_name in ['salt-minion', 'salt-minion-py3']:
if re.search(r'salt[\s_.-]*minion',
pkg_name,
flags=re.IGNORECASE + re.UNICODE) is not None:
ret[pkg_name] = {'install status': 'task started'}
if not __salt__['task.run'](name='update-salt-software'):
log.error('Failed to install {0}'.format(pkg_name))
@ -1360,7 +1362,8 @@ def install(name=None, refresh=False, pkgs=None, **kwargs):
else:
# Combine cmd and arguments
cmd = [cmd].extend(arguments)
cmd = [cmd]
cmd.extend(arguments)
# Launch the command
result = __salt__['cmd.run_all'](cmd,

4
salt/output/highstate.py
View File

@ -108,7 +108,7 @@ import pprint
import textwrap
# Import salt libs
import salt.utils
import salt.utils.color
import salt.utils.stringutils
import salt.output
from salt.utils.locales import sdecode
@ -158,7 +158,7 @@ def output(data, **kwargs): # pylint: disable=unused-argument
def _format_host(host, data):
host = sdecode(host)
colors = salt.utils.get_colors(
colors = salt.utils.color.get_colors(
__opts__.get('color'),
__opts__.get('color_theme'))
tabular = __opts__.get('state_tabular', False)

4
salt/output/key.py
View File

@ -8,9 +8,9 @@ The ``salt-key`` command makes use of this outputter to format its output.
from __future__ import absolute_import
# Import salt libs
import salt.utils
import salt.output
from salt.utils.locales import sdecode
import salt.utils.color
def output(data, **kwargs): # pylint: disable=unused-argument
@ -18,7 +18,7 @@ def output(data, **kwargs): # pylint: disable=unused-argument
Read in the dict structure generated by the salt key API methods and
print the structure.
'''
color = salt.utils.get_colors(
color = salt.utils.color.get_colors(
__opts__.get('color'),
__opts__.get('color_theme'))
strip_colors = __opts__.get('strip_colors', True)

4
salt/output/nested.py
View File

@ -29,9 +29,9 @@ from numbers import Number
# Import salt libs
import salt.output
import salt.utils.color
import salt.utils.locales
import salt.utils.odict
from salt.utils import get_colors
from salt.ext.six import string_types
@ -41,7 +41,7 @@ class NestDisplay(object):
'''
def __init__(self):
self.__dict__.update(
get_colors(
salt.utils.color.get_colors(
__opts__.get('color'),
__opts__.get('color_theme')
)

4
salt/output/no_return.py
View File

@ -15,7 +15,7 @@ Example output::
from __future__ import absolute_import
# Import salt libs
import salt.utils
import salt.utils.color
# Import 3rd-party libs
from salt.ext import six
@ -26,7 +26,7 @@ class NestDisplay(object):
Create generator for nested output
'''
def __init__(self):
self.colors = salt.utils.get_colors(
self.colors = salt.utils.color.get_colors(
__opts__.get(u'color'),
__opts__.get(u'color_theme'))

4
salt/output/overstatestage.py
View File

@ -11,7 +11,7 @@ and should not be called directly.
from __future__ import absolute_import
# Import Salt libs
import salt.utils
import salt.utils.color
# Import 3rd-party libs
from salt.ext import six
@ -27,7 +27,7 @@ def output(data, **kwargs): # pylint: disable=unused-argument
'''
Format the data for printing stage information from the overstate system
'''
colors = salt.utils.get_colors(
colors = salt.utils.color.get_colors(
__opts__.get('color'),
__opts__.get('color_theme'))
ostr = ''

10
salt/output/table_out.py
View File

@ -42,12 +42,10 @@ from functools import reduce # pylint: disable=redefined-builtin
# Import salt libs
import salt.output
import salt.utils.locales
from salt.ext.six import string_types
from salt.utils import get_colors
from salt.ext.six.moves import map # pylint: disable=redefined-builtin
from salt.ext.six.moves import zip # pylint: disable=redefined-builtin
from salt.ext.six.moves import map, zip # pylint: disable=redefined-builtin
import salt.utils.color
import salt.utils.locales
__virtualname__ = 'table'
@ -78,7 +76,7 @@ class TableDisplay(object):
width=50, # column max width
wrapfunc=None): # function wrapper
self.__dict__.update(
get_colors(
salt.utils.color.get_colors(
__opts__.get('color'),
__opts__.get('color_theme')
)

17
salt/states/git.py
View File

@ -1303,6 +1303,23 @@ def latest(name,
'if it does not already exist).',
comments
)
remote_tags = set([
x.replace('refs/tags/', '') for x in __salt__['git.ls_remote'](
cwd=target,
remote=remote,
opts="--tags",
user=user,
password=password,
identity=identity,
saltenv=__env__,
ignore_retcode=True,
).keys() if '^{}' not in x
])
if set(all_local_tags) != remote_tags:
has_remote_rev = False
ret['changes']['new_tags'] = list(remote_tags.symmetric_difference(
all_local_tags
))
if not has_remote_rev:
try:

124
salt/utils/__init__.py
View File

@ -126,7 +126,6 @@ import salt.utils.dictupdate
import salt.utils.versions
import salt.version
from salt.utils.decorators.jinja import jinja_filter
from salt.textformat import TextFormat
from salt.exceptions import (
CommandExecutionError, SaltClientError,
CommandNotFoundError, SaltSystemExit,
@ -138,83 +137,6 @@ log = logging.getLogger(__name__)
_empty = object()
def get_color_theme(theme):
'''
Return the color theme to use
'''
# Keep the heavy lifting out of the module space
import yaml
if not os.path.isfile(theme):
log.warning('The named theme {0} if not available'.format(theme))
# Late import to avoid circular import.
import salt.utils.files
try:
with salt.utils.files.fopen(theme, 'rb') as fp_:
colors = yaml.safe_load(fp_.read())
ret = {}
for color in colors:
ret[color] = '\033[{0}m'.format(colors[color])
if not isinstance(colors, dict):
log.warning('The theme file {0} is not a dict'.format(theme))
return {}
return ret
except Exception:
log.warning('Failed to read the color theme {0}'.format(theme))
return {}
def get_colors(use=True, theme=None):
'''
Return the colors as an easy to use dict. Pass `False` to deactivate all
colors by setting them to empty strings. Pass a string containing only the
name of a single color to be used in place of all colors. Examples:
.. code-block:: python
colors = get_colors() # enable all colors
no_colors = get_colors(False) # disable all colors
red_colors = get_colors('RED') # set all colors to red
'''
colors = {
'BLACK': TextFormat('black'),
'DARK_GRAY': TextFormat('bold', 'black'),
'RED': TextFormat('red'),
'LIGHT_RED': TextFormat('bold', 'red'),
'GREEN': TextFormat('green'),
'LIGHT_GREEN': TextFormat('bold', 'green'),
'YELLOW': TextFormat('yellow'),
'LIGHT_YELLOW': TextFormat('bold', 'yellow'),
'BLUE': TextFormat('blue'),
'LIGHT_BLUE': TextFormat('bold', 'blue'),
'MAGENTA': TextFormat('magenta'),
'LIGHT_MAGENTA': TextFormat('bold', 'magenta'),
'CYAN': TextFormat('cyan'),
'LIGHT_CYAN': TextFormat('bold', 'cyan'),
'LIGHT_GRAY': TextFormat('white'),
'WHITE': TextFormat('bold', 'white'),
'DEFAULT_COLOR': TextFormat('default'),
'ENDC': TextFormat('reset'),
}
if theme:
colors.update(get_color_theme(theme))
if not use:
for color in colors:
colors[color] = ''
if isinstance(use, six.string_types):
# Try to set all of the colors to the passed color
if use in colors:
for color in colors:
# except for color reset
if color == 'ENDC':
continue
colors[color] = colors[use]
return colors
def get_context(template, line, num_lines=5, marker=None):
'''
Returns debugging context around a line in a given string
@ -3435,3 +3357,49 @@ def kwargs_warn_until(kwargs,
stacklevel=stacklevel,
_version_info_=_version_info_,
_dont_call_warnings=_dont_call_warnings)
def get_color_theme(theme):
'''
Return the color theme to use
.. deprecated:: Oxygen
'''
# Late import to avoid circular import.
import salt.utils.color
import salt.utils.versions
salt.utils.versions.warn_until(
'Neon',
'Use of \'salt.utils.get_color_theme\' detected. This function has '
'been moved to \'salt.utils.color.get_color_theme\' as of Salt '
'Oxygen. This warning will be removed in Salt Neon.'
)
return salt.utils.color.get_color_theme(theme)
def get_colors(use=True, theme=None):
'''
Return the colors as an easy to use dict. Pass `False` to deactivate all
colors by setting them to empty strings. Pass a string containing only the
name of a single color to be used in place of all colors. Examples:
.. code-block:: python
colors = get_colors() # enable all colors
no_colors = get_colors(False) # disable all colors
red_colors = get_colors('RED') # set all colors to red
.. deprecated:: Oxygen
'''
# Late import to avoid circular import.
import salt.utils.color
import salt.utils.versions
salt.utils.versions.warn_until(
'Neon',
'Use of \'salt.utils.get_colors\' detected. This function has '
'been moved to \'salt.utils.color.get_colors\' as of Salt '
'Oxygen. This warning will be removed in Salt Neon.'
)
return salt.utils.color.get_colors(use=use, theme=theme)

92
salt/utils/color.py Normal file
View File

@ -0,0 +1,92 @@
# -*- coding: utf-8 -*-
'''
Functions used for CLI color themes.
'''
# Import Python libs
from __future__ import absolute_import
import logging
import os
# Import Salt libs
from salt.ext import six
from salt.textformat import TextFormat
import salt.utils.files
log = logging.getLogger(__name__)
def get_color_theme(theme):
'''
Return the color theme to use
'''
# Keep the heavy lifting out of the module space
import yaml
if not os.path.isfile(theme):
log.warning('The named theme {0} if not available'.format(theme))
try:
with salt.utils.files.fopen(theme, 'rb') as fp_:
colors = yaml.safe_load(fp_.read())
ret = {}
for color in colors:
ret[color] = '\033[{0}m'.format(colors[color])
if not isinstance(colors, dict):
log.warning('The theme file {0} is not a dict'.format(theme))
return {}
return ret
except Exception:
log.warning('Failed to read the color theme {0}'.format(theme))
return {}
def get_colors(use=True, theme=None):
'''
Return the colors as an easy to use dict. Pass `False` to deactivate all
colors by setting them to empty strings. Pass a string containing only the
name of a single color to be used in place of all colors. Examples:
.. code-block:: python
colors = get_colors() # enable all colors
no_colors = get_colors(False) # disable all colors
red_colors = get_colors('RED') # set all colors to red
'''
colors = {
'BLACK': TextFormat('black'),
'DARK_GRAY': TextFormat('bold', 'black'),
'RED': TextFormat('red'),
'LIGHT_RED': TextFormat('bold', 'red'),
'GREEN': TextFormat('green'),
'LIGHT_GREEN': TextFormat('bold', 'green'),
'YELLOW': TextFormat('yellow'),
'LIGHT_YELLOW': TextFormat('bold', 'yellow'),
'BLUE': TextFormat('blue'),
'LIGHT_BLUE': TextFormat('bold', 'blue'),
'MAGENTA': TextFormat('magenta'),
'LIGHT_MAGENTA': TextFormat('bold', 'magenta'),
'CYAN': TextFormat('cyan'),
'LIGHT_CYAN': TextFormat('bold', 'cyan'),
'LIGHT_GRAY': TextFormat('white'),
'WHITE': TextFormat('bold', 'white'),
'DEFAULT_COLOR': TextFormat('default'),
'ENDC': TextFormat('reset'),
}
if theme:
colors.update(get_color_theme(theme))
if not use:
for color in colors:
colors[color] = ''
if isinstance(use, six.string_types):
# Try to set all of the colors to the passed color
if use in colors:
for color in colors:
# except for color reset
if color == 'ENDC':
continue
colors[color] = colors[use]
return colors

4
tests/consist.py
View File

@ -9,13 +9,13 @@ import pprint
import optparse
# Import Salt libs
import salt.utils
import salt.utils.color
# Import 3rd-party libs
import yaml
from salt.ext import six
colors = salt.utils.get_colors()
colors = salt.utils.color.get_colors()
def parse():

5
tests/integration/__init__.py
View File

@ -49,7 +49,8 @@ import salt.minion
import salt.runner
import salt.output
import salt.version
import salt.utils # Can be removed once get_colors and appendproctitle are moved
import salt.utils # Can be removed once appendproctitle is moved
import salt.utils.color
import salt.utils.files
import salt.utils.path
import salt.utils.platform
@ -188,7 +189,7 @@ class TestDaemon(object):
def __init__(self, parser):
self.parser = parser
self.colors = salt.utils.get_colors(self.parser.options.no_colors is False)
self.colors = salt.utils.color.get_colors(self.parser.options.no_colors is False)
if salt.utils.platform.is_windows():
# There's no shell color support on windows...
for key in self.colors:

24
tests/unit/fileserver/test_gitfs.py
View File

@ -9,8 +9,12 @@ import os
import shutil
import tempfile
import textwrap
import pwd
import logging
import stat
try:
import pwd
except ImportError:
pass
# Import 3rd-party libs
import yaml
@ -210,7 +214,6 @@ class GitFSTest(TestCase, LoaderModuleMockMixin):
self.integration_base_files = os.path.join(FILES, 'file', 'base')
# Create the dir if it doesn't already exist
try:
shutil.copytree(self.integration_base_files, self.tmp_repo_dir + '/')
except OSError:
@ -224,7 +227,12 @@ class GitFSTest(TestCase, LoaderModuleMockMixin):
if 'USERNAME' not in os.environ:
try:
os.environ['USERNAME'] = pwd.getpwuid(os.geteuid()).pw_name
import salt.utils
if salt.utils.is_windows():
import salt.utils.win_functions
os.environ['USERNAME'] = salt.utils.win_functions.get_current_user()
else:
os.environ['USERNAME'] = pwd.getpwuid(os.geteuid()).pw_name
except AttributeError:
log.error('Unable to get effective username, falling back to '
'\'root\'.')
@ -240,14 +248,18 @@ class GitFSTest(TestCase, LoaderModuleMockMixin):
Remove the temporary git repository and gitfs cache directory to ensure
a clean environment for each test.
'''
shutil.rmtree(self.tmp_repo_dir)
shutil.rmtree(self.tmp_cachedir)
shutil.rmtree(self.tmp_sock_dir)
shutil.rmtree(self.tmp_repo_dir, onerror=self._rmtree_error)
shutil.rmtree(self.tmp_cachedir, onerror=self._rmtree_error)
shutil.rmtree(self.tmp_sock_dir, onerror=self._rmtree_error)
del self.tmp_repo_dir
del self.tmp_cachedir
del self.tmp_sock_dir
del self.integration_base_files
def _rmtree_error(self, func, path, excinfo):
os.chmod(path, stat.S_IWRITE)
func(path)
def test_file_list(self):
ret = gitfs.file_list(LOAD)
self.assertIn('testfile', ret)

53
tests/unit/modules/test_genesis.py
View File

@ -46,12 +46,57 @@ class GenesisTestCase(TestCase, LoaderModuleMockMixin):
with patch.dict(genesis.__salt__, {'disk.blkid': MagicMock(return_value={})}):
self.assertEqual(genesis.bootstrap('rpm', 'root', 'dir'), None)
with patch.object(genesis, '_bootstrap_deb', return_value='A'):
common_parms = {'platform': 'deb',
'root': 'root',
'img_format': 'dir',
'arch': 'amd64',
'flavor': 'stable',
'static_qemu': 'qemu'}
param_sets = [
{'params': {},
'cmd': ['debootstrap', '--foreign', '--arch', 'amd64',
'stable', 'root', 'http://ftp.debian.org/debian/']
},
{'params': {'pkgs': 'vim'},
'cmd': ['debootstrap', '--foreign', '--arch', 'amd64',
'--include', 'vim',
'stable', 'root', 'http://ftp.debian.org/debian/']
},
{'params': {'pkgs': 'vim,emacs'},
'cmd': ['debootstrap', '--foreign', '--arch', 'amd64',
'--include', 'vim,emacs',
'stable', 'root', 'http://ftp.debian.org/debian/']
},
{'params': {'pkgs': ['vim', 'emacs']},
'cmd': ['debootstrap', '--foreign', '--arch', 'amd64',
'--include', 'vim,emacs',
'stable', 'root', 'http://ftp.debian.org/debian/']
},
{'params': {'pkgs': ['vim', 'emacs'], 'exclude_pkgs': ['vim', 'foo']},
'cmd': ['debootstrap', '--foreign', '--arch', 'amd64',
'--include', 'vim,emacs', '--exclude', 'vim,foo',
'stable', 'root', 'http://ftp.debian.org/debian/']
},
]
for param_set in param_sets:
with patch.dict(genesis.__salt__, {'mount.umount': MagicMock(),
'file.rmdir': MagicMock(),
'file.directory_exists': MagicMock()}):
with patch.dict(genesis.__salt__, {'disk.blkid': MagicMock(return_value={})}):
self.assertEqual(genesis.bootstrap('deb', 'root', 'dir'), None)
'file.directory_exists': MagicMock(),
'cmd.run': MagicMock(),
'disk.blkid': MagicMock(return_value={})}):
with patch('salt.modules.genesis.salt.utils.which', return_value=True):
param_set['params'].update(common_parms)
self.assertEqual(genesis.bootstrap(**param_set['params']), None)
genesis.__salt__['cmd.run'].assert_any_call(param_set['cmd'], python_shell=False)
with patch.object(genesis, '_bootstrap_pacman', return_value='A') as pacman_patch:
with patch.dict(genesis.__salt__, {'mount.umount': MagicMock(),

7
tests/unit/modules/test_groupadd.py
View File

@ -5,7 +5,10 @@
# Import Python libs
from __future__ import absolute_import
import grp
try:
import grp
except ImportError:
pass
# Import Salt Testing Libs
from tests.support.mixins import LoaderModuleMockMixin
@ -13,10 +16,12 @@ from tests.support.unit import TestCase, skipIf
from tests.support.mock import MagicMock, patch, NO_MOCK, NO_MOCK_REASON
# Import Salt Libs
import salt.utils
import salt.modules.groupadd as groupadd
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(salt.utils.is_windows(), "Module not available on Windows")
class GroupAddTestCase(TestCase, LoaderModuleMockMixin):
'''
TestCase for salt.modules.groupadd

39
tests/unit/modules/test_inspect_collector.py
View File

@ -49,9 +49,15 @@ class InspectorCollectorTestCase(TestCase):
:return:
'''
inspector = Inspector(cachedir='/foo/cache', piddir='/foo/pid', pidfilename='bar.pid')
self.assertEqual(inspector.dbfile, '/foo/cache/_minion_collector.db')
self.assertEqual(inspector.pidfile, '/foo/pid/bar.pid')
cachedir = os.sep + os.sep.join(['foo', 'cache'])
piddir = os.sep + os.sep.join(['foo', 'pid'])
inspector = Inspector(cachedir=cachedir, piddir=piddir, pidfilename='bar.pid')
self.assertEqual(
inspector.dbfile,
os.sep + os.sep.join(['foo', 'cache', '_minion_collector.db']))
self.assertEqual(
inspector.pidfile,
os.sep + os.sep.join(['foo', 'pid', 'bar.pid']))
def test_file_tree(self):
'''
@ -60,12 +66,29 @@ class InspectorCollectorTestCase(TestCase):
:return:
'''
inspector = Inspector(cachedir='/test', piddir='/test', pidfilename='bar.pid')
inspector = Inspector(cachedir=os.sep + 'test',
piddir=os.sep + 'test',
pidfilename='bar.pid')
tree_root = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'inspectlib', 'tree_test')
expected_tree = (['/a/a/dummy.a', '/a/b/dummy.b', '/b/b.1', '/b/b.2', '/b/b.3'],
['/a', '/a/a', '/a/b', '/a/c', '/b', '/c'],
['/a/a/dummy.ln.a', '/a/b/dummy.ln.b', '/a/c/b.1', '/b/b.4',
'/b/b.5', '/c/b.1', '/c/b.2', '/c/b.3'])
expected_tree = ([os.sep + os.sep.join(['a', 'a', 'dummy.a']),
os.sep + os.sep.join(['a', 'b', 'dummy.b']),
os.sep + os.sep.join(['b', 'b.1']),
os.sep + os.sep.join(['b', 'b.2']),
os.sep + os.sep.join(['b', 'b.3'])],
[os.sep + 'a',
os.sep + os.sep.join(['a', 'a']),
os.sep + os.sep.join(['a', 'b']),
os.sep + os.sep.join(['a', 'c']),
os.sep + 'b',
os.sep + 'c'],
[os.sep + os.sep.join(['a', 'a', 'dummy.ln.a']),
os.sep + os.sep.join(['a', 'b', 'dummy.ln.b']),
os.sep + os.sep.join(['a', 'c', 'b.1']),
os.sep + os.sep.join(['b', 'b.4']),
os.sep + os.sep.join(['b', 'b.5']),
os.sep + os.sep.join(['c', 'b.1']),
os.sep + os.sep.join(['c', 'b.2']),
os.sep + os.sep.join(['c', 'b.3'])])
tree_result = []
for chunk in inspector._get_all_files(tree_root):
buff = []

9
tests/unit/modules/test_mac_group.py
View File

@ -5,11 +5,15 @@
# Import python libs
from __future__ import absolute_import
import grp
HAS_GRP = True
try:
import grp
except ImportError:
HAS_GRP = False
# Import Salt Testing Libs
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.unit import TestCase
from tests.support.unit import TestCase, skipIf
from tests.support.mock import MagicMock, patch
# Import Salt Libs
@ -17,6 +21,7 @@ import salt.modules.mac_group as mac_group
from salt.exceptions import SaltInvocationError, CommandExecutionError
@skipIf(not HAS_GRP, "Missing required library 'grp'")
class MacGroupTestCase(TestCase, LoaderModuleMockMixin):
'''
TestCase for the salt.modules.mac_group module

25
tests/unit/modules/test_mac_user.py
View File

@ -2,10 +2,13 @@
'''
:codeauthor: :email:`Nicole Thomas <nicole@saltstack.com>`
'''
# Import python libs
from __future__ import absolute_import
import pwd
HAS_PWD = True
try:
import pwd
except ImportError:
HAS_PWD = False
# Import Salt Testing Libs
from tests.support.mixins import LoaderModuleMockMixin
@ -17,6 +20,7 @@ import salt.modules.mac_user as mac_user
from salt.exceptions import SaltInvocationError, CommandExecutionError
@skipIf(not HAS_PWD, "Missing required library 'pwd'")
@skipIf(NO_MOCK, NO_MOCK_REASON)
class MacUserTestCase(TestCase, LoaderModuleMockMixin):
'''
@ -26,14 +30,15 @@ class MacUserTestCase(TestCase, LoaderModuleMockMixin):
def setup_loader_modules(self):
return {mac_user: {}}
mock_pwall = [pwd.struct_passwd(('_amavisd', '*', 83, 83, 'AMaViS Daemon',
'/var/virusmails', '/usr/bin/false')),
pwd.struct_passwd(('_appleevents', '*', 55, 55,
'AppleEvents Daemon',
'/var/empty', '/usr/bin/false')),
pwd.struct_passwd(('_appowner', '*', 87, 87,
'Application Owner',
'/var/empty', '/usr/bin/false'))]
if HAS_PWD:
mock_pwall = [pwd.struct_passwd(('_amavisd', '*', 83, 83, 'AMaViS Daemon',
'/var/virusmails', '/usr/bin/false')),
pwd.struct_passwd(('_appleevents', '*', 55, 55,
'AppleEvents Daemon',
'/var/empty', '/usr/bin/false')),
pwd.struct_passwd(('_appowner', '*', 87, 87,
'Application Owner',
'/var/empty', '/usr/bin/false'))]
mock_info_ret = {'shell': '/bin/bash', 'name': 'test', 'gid': 4376,
'groups': ['TEST_GROUP'], 'home': '/Users/foo',
'fullname': 'TEST USER', 'uid': 4376}

324
tests/unit/modules/test_saltcheck.py Normal file
View File

@ -0,0 +1,324 @@
# -*- coding: utf-8 -*-
'''Unit test for saltcheck execution module'''
# Import Python libs
from __future__ import absolute_import
try:
import salt.modules.saltcheck as saltcheck
except:
raise
# Import Salt Testing Libs
try:
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.unit import skipIf, TestCase
from tests.support.mock import (
MagicMock,
patch,
NO_MOCK,
NO_MOCK_REASON
)
except:
raise
@skipIf(NO_MOCK, NO_MOCK_REASON)
class LinuxSysctlTestCase(TestCase, LoaderModuleMockMixin):
'''
TestCase for salt.modules.saltcheck module
'''
def setup_loader_modules(self):
return {saltcheck: {}}
def test_call_salt_command(self):
'''test simple test.echo module'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'sys.list_modules': MagicMock(return_value=['module1']),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
returned = sc_instance.call_salt_command(fun="test.echo", args=['hello'], kwargs=None)
self.assertEqual(returned, 'hello')
def test_update_master_cache(self):
'''test master cache'''
self.assertTrue(saltcheck.update_master_cache)
def test_call_salt_command2(self):
'''test simple test.echo module again'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'sys.list_modules': MagicMock(return_value=['module1']),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
returned = sc_instance.call_salt_command(fun="test.echo", args=['hello'], kwargs=None)
self.assertNotEqual(returned, 'not-hello')
def test__assert_equal1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = {'a': 1, 'b': 2}
bbb = {'a': 1, 'b': 2}
mybool = sc_instance._SaltCheck__assert_equal(aaa, bbb)
self.assertTrue(mybool)
def test__assert_equal2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
mybool = sc_instance._SaltCheck__assert_equal(False, True)
self.assertNotEqual(mybool, True)
def test__assert_not_equal1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = {'a': 1, 'b': 2}
bbb = {'a': 1, 'b': 2, 'c': 3}
mybool = sc_instance._SaltCheck__assert_not_equal(aaa, bbb)
self.assertTrue(mybool)
def test__assert_not_equal2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = {'a': 1, 'b': 2}
bbb = {'a': 1, 'b': 2}
mybool = sc_instance._SaltCheck__assert_not_equal(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_true1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
mybool = sc_instance._SaltCheck__assert_equal(True, True)
self.assertTrue(mybool)
def test__assert_true2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
mybool = sc_instance._SaltCheck__assert_equal(False, True)
self.assertNotEqual(mybool, True)
def test__assert_false1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
mybool = sc_instance._SaltCheck__assert_false(False)
self.assertTrue(mybool)
def test__assert_false2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
mybool = sc_instance._SaltCheck__assert_false(True)
self.assertNotEqual(mybool, True)
def test__assert_in1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = "bob"
mylist = ['alice', 'bob', 'charles', 'dana']
mybool = sc_instance._SaltCheck__assert_in(aaa, mylist)
self.assertTrue(mybool, True)
def test__assert_in2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = "elaine"
mylist = ['alice', 'bob', 'charles', 'dana']
mybool = sc_instance._SaltCheck__assert_in(aaa, mylist)
self.assertNotEqual(mybool, True)
def test__assert_not_in1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = "elaine"
mylist = ['alice', 'bob', 'charles', 'dana']
mybool = sc_instance._SaltCheck__assert_not_in(aaa, mylist)
self.assertTrue(mybool, True)
def test__assert_not_in2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = "bob"
mylist = ['alice', 'bob', 'charles', 'dana']
mybool = sc_instance._SaltCheck__assert_not_in(aaa, mylist)
self.assertNotEqual(mybool, True)
def test__assert_greater1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 110
bbb = 100
mybool = sc_instance._SaltCheck__assert_greater(aaa, bbb)
self.assertTrue(mybool, True)
def test__assert_greater2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 110
mybool = sc_instance._SaltCheck__assert_greater(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_greater3(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 100
mybool = sc_instance._SaltCheck__assert_greater(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_greater_equal1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 110
bbb = 100
mybool = sc_instance._SaltCheck__assert_greater_equal(aaa, bbb)
self.assertTrue(mybool, True)
def test__assert_greater_equal2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 110
mybool = sc_instance._SaltCheck__assert_greater_equal(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_greater_equal3(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 100
mybool = sc_instance._SaltCheck__assert_greater_equal(aaa, bbb)
self.assertEqual(mybool, 'Pass')
def test__assert_less1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 99
bbb = 100
mybool = sc_instance._SaltCheck__assert_less(aaa, bbb)
self.assertTrue(mybool, True)
def test__assert_less2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 110
bbb = 99
mybool = sc_instance._SaltCheck__assert_less(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_less3(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 100
mybool = sc_instance._SaltCheck__assert_less(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_less_equal1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 99
bbb = 100
mybool = sc_instance._SaltCheck__assert_less_equal(aaa, bbb)
self.assertTrue(mybool, True)
def test__assert_less_equal2(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 110
bbb = 99
mybool = sc_instance._SaltCheck__assert_less_equal(aaa, bbb)
self.assertNotEqual(mybool, True)
def test__assert_less_equal3(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'cp.cache_master': MagicMock(return_value=[True])
}):
sc_instance = saltcheck.SaltCheck()
aaa = 100
bbb = 100
mybool = sc_instance._SaltCheck__assert_less_equal(aaa, bbb)
self.assertEqual(mybool, 'Pass')
def test_run_test_1(self):
'''test'''
with patch.dict(saltcheck.__salt__, {'config.get': MagicMock(return_value=True),
'sys.list_modules': MagicMock(return_value=['test']),
'sys.list_functions': MagicMock(return_value=['test.echo']),
'cp.cache_master': MagicMock(return_value=[True])}):
returned = saltcheck.run_test(test={"module_and_function": "test.echo",
"assertion": "assertEqual",
"expected-return": "This works!",
"args": ["This works!"]
})
self.assertEqual(returned, 'Pass')

27
tests/unit/utils/test_color.py Normal file
View File

@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
'''
Unit tests for salt.utils.color.py
'''
# Import python libs
from __future__ import absolute_import
# Import Salt Testing libs
from tests.support.unit import TestCase
# Import Salt libs
import salt.utils.color
class ColorUtilsTestCase(TestCase):
def test_get_colors(self):
ret = salt.utils.color.get_colors()
self.assertEqual('\x1b[0;37m', str(ret['LIGHT_GRAY']))
ret = salt.utils.color.get_colors(use=False)
self.assertDictContainsSubset({'LIGHT_GRAY': ''}, ret)
ret = salt.utils.color.get_colors(use='LIGHT_GRAY')
# LIGHT_YELLOW now == LIGHT_GRAY
self.assertEqual(str(ret['LIGHT_YELLOW']), str(ret['LIGHT_GRAY']))

11
tests/unit/utils/test_utils.py
View File

@ -895,17 +895,6 @@ class UtilsTestCase(TestCase):
ret = salt.utils.repack_dictlist(LOREM_IPSUM)
self.assertDictEqual(ret, {})
def test_get_colors(self):
ret = salt.utils.get_colors()
self.assertEqual('\x1b[0;37m', str(ret['LIGHT_GRAY']))
ret = salt.utils.get_colors(use=False)
self.assertDictContainsSubset({'LIGHT_GRAY': ''}, ret)
ret = salt.utils.get_colors(use='LIGHT_GRAY')
# LIGHT_YELLOW now == LIGHT_GRAY
self.assertEqual(str(ret['LIGHT_YELLOW']), str(ret['LIGHT_GRAY']))
@skipIf(NO_MOCK, NO_MOCK_REASON)
def test_daemonize_if(self):
# pylint: disable=assignment-from-none