Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_obj_serialization(self):
'''
Test object serialization.
:return:
'''
obj = FoobarEntity()
obj.foo = 123
obj.bar = 'test entity'
obj.spam = 0.123
descr = OrderedDict([tuple(elm.split(':')) for elm in ["foo:int", "bar:str", "spam:float"]])
assert obj._serialize(descr) == [123, 'test entity', 0.123]
def test_snapshot_failure(self):
'''
Tests zfs snapshot failure
'''
res = OrderedDict([
('snapshotted', False),
('error', "cannot create snapshot 'myzpool/mydataset@baseline': dataset already exists"),
])
ret = {'pid': 68526, 'retcode': 1, 'stderr': "cannot create snapshot 'myzpool/mydataset@baseline': dataset already exists", 'stdout': ''}
mock_cmd = MagicMock(return_value=ret)
with patch.dict(zfs.__salt__, {'cmd.run_all': mock_cmd}), \
patch.dict(zfs.__utils__, self.utils_patch):
self.assertEqual(res, zfs.snapshot('myzpool/mydataset@baseline'))
]),
{
'test_ivstate0': {
'__env__': 'base',
'__sls__': u'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}],
't': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
'run',
{'order': 10002}]},
'test_ivstate1': {
'__env__': 'base',
'__sls__': u'a',
'cmd': [salt.utils.odict.OrderedDict([('name', '/bin/true')]),
salt.utils.odict.OrderedDict([
('onfail_stop', False),
('onfail',
[salt.utils.odict.OrderedDict([('cmd', 'test_ivstate0')])])
]),
'run',
{'order': 10004}]},
}
),
'test_onfail_integ4': (
salt.utils.odict.OrderedDict([
('host1',
salt.utils.odict.OrderedDict([
('t_|-test_ivstate0_|-echo_|-run', {
'result': False}),
('cmd_|-test_ivstate0_|-echo_|-run', {
'result': False}),
def default_reverse_proxy(target):
data = OrderedDict([('target', target),
('sw_proxies', []),
('network_mappings', OrderedDict()),
('http_proxy', OrderedDict()),
('http_backends', OrderedDict()),
('https_proxy', OrderedDict()),
('https_backends', OrderedDict())])
return data
'_settings': 'settings',
'_metadata': 'metadata'}
_CACHEKEY = 'localreg_'
_DEFAULT_MC = 'default'
_LOCAL_CACHES = {}
_DEFAULT_KEY = 'cache_key_{0}'
_default = object()
RE_FLAGS = re.M | re.U | re.S
STRIP_FLAGS = RE_FLAGS
_CACHE_KEYS = {}
STRIPPED_RES = [
re.compile(r"\x1b\[[0-9;]*[mG]", STRIP_FLAGS),
re.compile(r"\x1b.*?[mGKH]", STRIP_FLAGS)]
_USE_MEMCACHE_FIRST = OrderedDict()
_USE_MEMOIZE_FIRST = OrderedDict()
log = logging.getLogger('mc_states.api')
OKEYS = {}
_CONFIG_CACHE_KEYS = ['master',
'config_dir',
'local',
'conf_file',
'master_port',
'file_roots',
'pillar_roots',
'__role',
'user',
'username',
'conf_prefix',
'publish_port',
'id',
'ret_port']
def default_settings():
_s = __salt__
data = {'controller': _s['mc_pillar.minion_id'](),
'vts': {'generic': True,
'saltify': True,
'lxc': False,
'kvm': False},
'compute_nodes': OrderedDict(),
'vms': OrderedDict()}
return data
def create_file_vdev(size, *vdevs):
'''
Creates file based virtual devices for a zpool
CLI Example:
.. code-block:: bash
salt '*' zpool.create_file_vdev 7G /path/to/vdev1 [/path/to/vdev2] [...]
.. note::
Depending on file size, the above command may take a while to return.
'''
ret = OrderedDict()
err = OrderedDict()
_mkfile_cmd = salt.utils.path.which('mkfile')
for vdev in vdevs:
if os.path.isfile(vdev):
ret[vdev] = 'existed'
else:
res = __salt__['cmd.run_all'](
'{mkfile} {size} {vdev}'.format(
mkfile=_mkfile_cmd,
size=size,
vdev=vdev,
),
python_shell=False,
)
if res['retcode'] != 0:
def _do():
data = OrderedDict()
for cn, cdata in get_cloud_conf_by_cns().items():
cvms = cdata.pop('vms')
for vt in cdata['vts']:
vtdata = data.setdefault(vt, OrderedDict())
vcndata = vtdata.setdefault(cn, OrderedDict())
vcnvms = vcndata.setdefault('vms', OrderedDict())
for vm, vmdata in cvms.items():
vt = vmdata['vt']
vtdata = data.setdefault(vt, OrderedDict())
vcndata = vtdata.setdefault(cn, copy.deepcopy(cdata))
vcnvms = vcndata.setdefault('vms', OrderedDict())
vcnvms[vm] = vmdata
return data
cache_key = __name + '.get_cloud_conf_by_vts'
- setup reverse proxying
- setup it's local internal addressing dns to point to private ips
- everything else that's local to the compute node
The computes nodes are often created implicitly by registration of vms
on specific drivers like LXC but you can get_cloud_confr some manually.
makina-states.cloud.compute_node.settings.targets.devhost11.local: {}
To add or modify a value, use the mc_utils.default habitual way of
modifying the default dict.
'''
target = __grains__['id']
data = {'has': {'firewall': True},
'target': target,
'vms': OrderedDict(),
'expose': [],
'expose_limited': {},
'domains': [],
'vts': [],
'http_port': 80,
'https_port': 443,
'reverse_proxies': default_reverse_proxy(target),
'excluded_ports': [],
'port_range_start': 40000,
'port_range_end': 60000}
data.update(__salt__['mc_cloud.ssh_settings']())
return data
def _dogather_expositions():
_s = __salt__
data = OrderedDict()
direct = data.setdefault('direct', OrderedDict())
indirect = data.setdefault('indirect', OrderedDict())
vms = _s['mc_cloud_compute_node.get_vms']()
targets = _s['mc_cloud_compute_node.get_targets']()
for vkind, fun, mapping in [
('vms',
'mc_pillar.get_cloud_conf_for_vm',
vms),
('cns',
'mc_pillar.get_cloud_conf_for_cn',
targets),
]:
for i, tdata in six.iteritems(mapping):
if _s['mc_cloud.is_a_compute_node'](i):
kind = 'cns'
else: