mirror of
https://github.com/valitydev/salt.git
synced 2024-11-07 17:09:03 +00:00
879c557264
This adds a new decorator which creates a temporary directory and cleans it after the test completes. It also modifies an existing decorator for creating temporary files so that it accepts arguments, which will be passed through to salt.utils.files.mkstemp().
425 lines
15 KiB
Python
425 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Import Python libs
|
|
from __future__ import absolute_import
|
|
import os
|
|
import sys
|
|
import shutil
|
|
import tempfile
|
|
import textwrap
|
|
import copy
|
|
|
|
# Import Salt Testing libs
|
|
from tests.support.helpers import with_tempdir
|
|
from tests.support.unit import TestCase
|
|
from tests.support.paths import TMP
|
|
|
|
# Import Salt libs
|
|
import salt.loader
|
|
import salt.config
|
|
import salt.utils.files
|
|
import salt.utils.versions
|
|
from salt.state import HighState
|
|
from salt.utils.pydsl import PyDslError
|
|
|
|
|
|
# Import 3rd-party libs
|
|
from salt.ext import six
|
|
from salt.ext.six.moves import StringIO
|
|
|
|
|
|
REQUISITES = ['require', 'require_in', 'use', 'use_in', 'watch', 'watch_in']
|
|
|
|
|
|
class CommonTestCaseBoilerplate(TestCase):
|
|
|
|
def setUp(self):
|
|
self.root_dir = tempfile.mkdtemp(dir=TMP)
|
|
self.addCleanup(shutil.rmtree, self.root_dir, ignore_errors=True)
|
|
self.state_tree_dir = os.path.join(self.root_dir, 'state_tree')
|
|
self.cache_dir = os.path.join(self.root_dir, 'cachedir')
|
|
if not os.path.isdir(self.root_dir):
|
|
os.makedirs(self.root_dir)
|
|
|
|
if not os.path.isdir(self.state_tree_dir):
|
|
os.makedirs(self.state_tree_dir)
|
|
|
|
if not os.path.isdir(self.cache_dir):
|
|
os.makedirs(self.cache_dir)
|
|
self.config = salt.config.minion_config(None)
|
|
self.config['root_dir'] = self.root_dir
|
|
self.config['state_events'] = False
|
|
self.config['id'] = 'match'
|
|
self.config['file_client'] = 'local'
|
|
self.config['file_roots'] = dict(base=[self.state_tree_dir])
|
|
self.config['cachedir'] = self.cache_dir
|
|
self.config['test'] = False
|
|
self.config['grains'] = salt.loader.grains(self.config)
|
|
self.HIGHSTATE = HighState(self.config)
|
|
self.HIGHSTATE.push_active()
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.HIGHSTATE.pop_active()
|
|
except IndexError:
|
|
pass
|
|
del self.config
|
|
del self.HIGHSTATE
|
|
|
|
def state_highstate(self, state, dirpath):
|
|
opts = copy.copy(self.config)
|
|
opts['file_roots'] = dict(base=[dirpath])
|
|
HIGHSTATE = HighState(opts)
|
|
HIGHSTATE.push_active()
|
|
try:
|
|
high, errors = HIGHSTATE.render_highstate(state)
|
|
if errors:
|
|
import pprint
|
|
pprint.pprint('\n'.join(errors))
|
|
pprint.pprint(high)
|
|
|
|
out = HIGHSTATE.state.call_high(high)
|
|
# pprint.pprint(out)
|
|
finally:
|
|
HIGHSTATE.pop_active()
|
|
|
|
|
|
class PyDSLRendererTestCase(CommonTestCaseBoilerplate):
|
|
'''
|
|
WARNING: If tests in here are flaky, they may need
|
|
to be moved to their own class. Sharing HighState, especially
|
|
through setUp/tearDown can create dangerous race conditions!
|
|
'''
|
|
|
|
def render_sls(self, content, sls='', saltenv='base', **kws):
|
|
if 'env' in kws:
|
|
# "env" is not supported; Use "saltenv".
|
|
kws.pop('env')
|
|
|
|
return self.HIGHSTATE.state.rend['pydsl'](
|
|
StringIO(content), saltenv=saltenv, sls=sls, **kws
|
|
)
|
|
|
|
def test_state_declarations(self):
|
|
result = self.render_sls(textwrap.dedent('''
|
|
state('A').cmd.run('ls -la', cwd='/var/tmp')
|
|
state().file.managed('myfile.txt', source='salt://path/to/file')
|
|
state('X').cmd('run', 'echo hello world', cwd='/')
|
|
|
|
a_cmd = state('A').cmd
|
|
a_cmd.run(shell='/bin/bash')
|
|
state('A').service.running(name='apache')
|
|
'''))
|
|
self.assertTrue('A' in result and 'X' in result)
|
|
A_cmd = result['A']['cmd']
|
|
self.assertEqual(A_cmd[0], 'run')
|
|
self.assertEqual(A_cmd[1]['name'], 'ls -la')
|
|
self.assertEqual(A_cmd[2]['cwd'], '/var/tmp')
|
|
self.assertEqual(A_cmd[3]['shell'], '/bin/bash')
|
|
|
|
A_service = result['A']['service']
|
|
self.assertEqual(A_service[0], 'running')
|
|
self.assertEqual(A_service[1]['name'], 'apache')
|
|
|
|
X_cmd = result['X']['cmd']
|
|
self.assertEqual(X_cmd[0], 'run')
|
|
self.assertEqual(X_cmd[1]['name'], 'echo hello world')
|
|
self.assertEqual(X_cmd[2]['cwd'], '/')
|
|
|
|
del result['A']
|
|
del result['X']
|
|
self.assertEqual(len(result), 2)
|
|
# 2 rather than 1 because pydsl adds an extra no-op state
|
|
# declaration.
|
|
|
|
s_iter = six.itervalues(result)
|
|
try:
|
|
s = next(s_iter)['file']
|
|
except KeyError:
|
|
s = next(s_iter)['file']
|
|
self.assertEqual(s[0], 'managed')
|
|
self.assertEqual(s[1]['name'], 'myfile.txt')
|
|
self.assertEqual(s[2]['source'], 'salt://path/to/file')
|
|
|
|
def test_requisite_declarations(self):
|
|
result = self.render_sls(textwrap.dedent('''
|
|
state('X').cmd.run('echo hello')
|
|
state('A').cmd.run('mkdir tmp', cwd='/var')
|
|
state('B').cmd.run('ls -la', cwd='/var/tmp') \
|
|
.require(state('X').cmd) \
|
|
.require(cmd='A') \
|
|
.watch(service='G')
|
|
state('G').service.running(name='collectd')
|
|
state('G').service.watch_in(state('A').cmd)
|
|
|
|
state('H').cmd.require_in(cmd='echo hello')
|
|
state('H').cmd.run('echo world')
|
|
'''))
|
|
self.assertTrue(len(result), 6)
|
|
self.assertTrue(set("X A B G H".split()).issubset(set(result.keys())))
|
|
b = result['B']['cmd']
|
|
self.assertEqual(b[0], 'run')
|
|
self.assertEqual(b[1]['name'], 'ls -la')
|
|
self.assertEqual(b[2]['cwd'], '/var/tmp')
|
|
self.assertEqual(b[3]['require'][0]['cmd'], 'X')
|
|
self.assertEqual(b[4]['require'][0]['cmd'], 'A')
|
|
self.assertEqual(b[5]['watch'][0]['service'], 'G')
|
|
self.assertEqual(result['G']['service'][2]['watch_in'][0]['cmd'], 'A')
|
|
self.assertEqual(
|
|
result['H']['cmd'][1]['require_in'][0]['cmd'], 'echo hello'
|
|
)
|
|
|
|
def test_include_extend(self):
|
|
result = self.render_sls(textwrap.dedent('''
|
|
include(
|
|
'some.sls.file',
|
|
'another.sls.file',
|
|
'more.sls.file',
|
|
delayed=True
|
|
)
|
|
A = state('A').cmd.run('echo hoho', cwd='/')
|
|
state('B').cmd.run('echo hehe', cwd='/')
|
|
extend(
|
|
A,
|
|
state('X').cmd.run(cwd='/a/b/c'),
|
|
state('Y').file('managed', name='a_file.txt'),
|
|
state('Z').service.watch(file='A')
|
|
)
|
|
'''))
|
|
self.assertEqual(len(result), 4)
|
|
self.assertEqual(
|
|
result['include'],
|
|
[{'base': sls} for sls in
|
|
('some.sls.file', 'another.sls.file', 'more.sls.file')]
|
|
)
|
|
extend = result['extend']
|
|
self.assertEqual(extend['X']['cmd'][0], 'run')
|
|
self.assertEqual(extend['X']['cmd'][1]['cwd'], '/a/b/c')
|
|
self.assertEqual(extend['Y']['file'][0], 'managed')
|
|
self.assertEqual(extend['Y']['file'][1]['name'], 'a_file.txt')
|
|
self.assertEqual(len(extend['Z']['service']), 1)
|
|
self.assertEqual(extend['Z']['service'][0]['watch'][0]['file'], 'A')
|
|
|
|
self.assertEqual(result['B']['cmd'][0], 'run')
|
|
self.assertTrue('A' not in result)
|
|
self.assertEqual(extend['A']['cmd'][0], 'run')
|
|
|
|
def test_cmd_call(self):
|
|
result = self.HIGHSTATE.state.call_template_str(textwrap.dedent('''\
|
|
#!pydsl
|
|
state('A').cmd.run('echo this is state A', cwd='/')
|
|
|
|
some_var = 12345
|
|
def do_something(a, b, *args, **kws):
|
|
return dict(result=True, changes={'a': a, 'b': b, 'args': args, 'kws': kws, 'some_var': some_var})
|
|
|
|
state('C').cmd.call(do_something, 1, 2, 3, x=1, y=2) \
|
|
.require(state('A').cmd)
|
|
|
|
state('G').cmd.wait('echo this is state G', cwd='/') \
|
|
.watch(state('C').cmd)
|
|
'''))
|
|
ret = next(result[k] for k in six.iterkeys(result) if 'do_something' in k)
|
|
changes = ret['changes']
|
|
self.assertEqual(
|
|
changes,
|
|
dict(a=1, b=2, args=(3,), kws=dict(x=1, y=2), some_var=12345)
|
|
)
|
|
|
|
ret = next(result[k] for k in six.iterkeys(result) if '-G_' in k)
|
|
self.assertEqual(ret['changes']['stdout'], 'this is state G')
|
|
|
|
def test_multiple_state_func_in_state_mod(self):
|
|
with self.assertRaisesRegex(PyDslError, 'Multiple state functions'):
|
|
self.render_sls(textwrap.dedent('''
|
|
state('A').cmd.run('echo hoho')
|
|
state('A').cmd.wait('echo hehe')
|
|
'''))
|
|
|
|
def test_no_state_func_in_state_mod(self):
|
|
with self.assertRaisesRegex(PyDslError, 'No state function specified'):
|
|
self.render_sls(textwrap.dedent('''
|
|
state('B').cmd.require(cmd='hoho')
|
|
'''))
|
|
|
|
def test_load_highstate(self):
|
|
result = self.render_sls(textwrap.dedent('''
|
|
import salt.utils.yaml
|
|
__pydsl__.load_highstate(salt.utils.yaml.safe_load("""
|
|
A:
|
|
cmd.run:
|
|
- name: echo hello
|
|
- cwd: /
|
|
B:
|
|
pkg:
|
|
- installed
|
|
service:
|
|
- running
|
|
- require:
|
|
- pkg: B
|
|
- watch:
|
|
- cmd: A
|
|
"""))
|
|
|
|
state('A').cmd.run(name='echo hello world')
|
|
'''))
|
|
self.assertEqual(len(result), 3)
|
|
self.assertEqual(result['A']['cmd'][0], 'run')
|
|
self.assertIn({'name': 'echo hello'}, result['A']['cmd'])
|
|
self.assertIn({'cwd': '/'}, result['A']['cmd'])
|
|
self.assertIn({'name': 'echo hello world'}, result['A']['cmd'])
|
|
self.assertEqual(len(result['A']['cmd']), 4)
|
|
|
|
self.assertEqual(len(result['B']['pkg']), 1)
|
|
self.assertEqual(result['B']['pkg'][0], 'installed')
|
|
|
|
self.assertEqual(result['B']['service'][0], 'running')
|
|
self.assertIn({'require': [{'pkg': 'B'}]}, result['B']['service'])
|
|
self.assertIn({'watch': [{'cmd': 'A'}]}, result['B']['service'])
|
|
self.assertEqual(len(result['B']['service']), 3)
|
|
|
|
def test_ordered_states(self):
|
|
result = self.render_sls(textwrap.dedent('''
|
|
__pydsl__.set(ordered=True)
|
|
A = state('A')
|
|
state('B').cmd.run('echo bbbb')
|
|
A.cmd.run('echo aaa')
|
|
state('B').cmd.run(cwd='/')
|
|
state('C').cmd.run('echo ccc')
|
|
state('B').file.managed(source='/a/b/c')
|
|
'''))
|
|
self.assertEqual(len(result['B']['cmd']), 3)
|
|
self.assertEqual(result['A']['cmd'][1]['require'][0]['cmd'], 'B')
|
|
self.assertEqual(result['C']['cmd'][1]['require'][0]['cmd'], 'A')
|
|
self.assertEqual(result['B']['file'][1]['require'][0]['cmd'], 'C')
|
|
|
|
@with_tempdir()
|
|
def test_pipe_through_stateconf(self, dirpath):
|
|
output = os.path.join(dirpath, 'output')
|
|
write_to(os.path.join(dirpath, 'xxx.sls'), textwrap.dedent(
|
|
'''#!stateconf -os yaml . jinja
|
|
.X:
|
|
cmd.run:
|
|
- name: echo X >> {0}
|
|
- cwd: /
|
|
.Y:
|
|
cmd.run:
|
|
- name: echo Y >> {0}
|
|
- cwd: /
|
|
.Z:
|
|
cmd.run:
|
|
- name: echo Z >> {0}
|
|
- cwd: /
|
|
'''.format(output.replace('\\', '/'))))
|
|
write_to(os.path.join(dirpath, 'yyy.sls'), textwrap.dedent('''\
|
|
#!pydsl|stateconf -ps
|
|
|
|
__pydsl__.set(ordered=True)
|
|
state('.D').cmd.run('echo D >> {0}', cwd='/')
|
|
state('.E').cmd.run('echo E >> {0}', cwd='/')
|
|
state('.F').cmd.run('echo F >> {0}', cwd='/')
|
|
'''.format(output.replace('\\', '/'))))
|
|
|
|
write_to(os.path.join(dirpath, 'aaa.sls'), textwrap.dedent('''\
|
|
#!pydsl|stateconf -ps
|
|
|
|
include('xxx', 'yyy')
|
|
|
|
# make all states in xxx run BEFORE states in this sls.
|
|
extend(state('.start').stateconf.require(stateconf='xxx::goal'))
|
|
|
|
# make all states in yyy run AFTER this sls.
|
|
extend(state('.goal').stateconf.require_in(stateconf='yyy::start'))
|
|
|
|
__pydsl__.set(ordered=True)
|
|
|
|
state('.A').cmd.run('echo A >> {0}', cwd='/')
|
|
state('.B').cmd.run('echo B >> {0}', cwd='/')
|
|
state('.C').cmd.run('echo C >> {0}', cwd='/')
|
|
'''.format(output.replace('\\', '/'))))
|
|
|
|
self.state_highstate({'base': ['aaa']}, dirpath)
|
|
with salt.utils.files.fopen(output, 'r') as f:
|
|
self.assertEqual(''.join(f.read().split()), "XYZABCDEF")
|
|
|
|
@with_tempdir()
|
|
def test_compile_time_state_execution(self, dirpath):
|
|
if not sys.stdin.isatty():
|
|
self.skipTest('Not attached to a TTY')
|
|
# The Windows shell will include any spaces before the redirect
|
|
# in the text that is redirected.
|
|
# For example: echo hello > test.txt will contain "hello "
|
|
write_to(os.path.join(dirpath, 'aaa.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
|
|
__pydsl__.set(ordered=True)
|
|
A = state('A')
|
|
A.cmd.run('echo hehe>{0}/zzz.txt', cwd='/')
|
|
A.file.managed('{0}/yyy.txt', source='salt://zzz.txt')
|
|
A()
|
|
A()
|
|
|
|
state().cmd.run('echo hoho>>{0}/yyy.txt', cwd='/')
|
|
|
|
A.file.managed('{0}/xxx.txt', source='salt://zzz.txt')
|
|
A()
|
|
'''.format(dirpath.replace('\\', '/'))))
|
|
self.state_highstate({'base': ['aaa']}, dirpath)
|
|
with salt.utils.files.fopen(os.path.join(dirpath, 'yyy.txt'), 'rt') as f:
|
|
self.assertEqual(f.read(), 'hehe' + os.linesep + 'hoho' + os.linesep)
|
|
with salt.utils.files.fopen(os.path.join(dirpath, 'xxx.txt'), 'rt') as f:
|
|
self.assertEqual(f.read(), 'hehe' + os.linesep)
|
|
|
|
@with_tempdir()
|
|
def test_nested_high_state_execution(self, dirpath):
|
|
output = os.path.join(dirpath, 'output')
|
|
write_to(os.path.join(dirpath, 'aaa.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
__salt__['state.sls']('bbb')
|
|
state().cmd.run('echo bbbbbb', cwd='/')
|
|
'''))
|
|
write_to(os.path.join(dirpath, 'bbb.sls'), textwrap.dedent(
|
|
'''
|
|
# {{ salt['state.sls']('ccc') }}
|
|
test:
|
|
cmd.run:
|
|
- name: echo bbbbbbb
|
|
- cwd: /
|
|
'''))
|
|
write_to(os.path.join(dirpath, 'ccc.sls'), textwrap.dedent(
|
|
'''
|
|
#!pydsl
|
|
state().cmd.run('echo ccccc', cwd='/')
|
|
'''))
|
|
self.state_highstate({'base': ['aaa']}, dirpath)
|
|
|
|
@with_tempdir()
|
|
def test_repeat_includes(self, dirpath):
|
|
output = os.path.join(dirpath, 'output')
|
|
write_to(os.path.join(dirpath, 'b.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
include('c')
|
|
include('d')
|
|
'''))
|
|
write_to(os.path.join(dirpath, 'c.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
modtest = include('e')
|
|
modtest.success
|
|
'''))
|
|
write_to(os.path.join(dirpath, 'd.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
modtest = include('e')
|
|
modtest.success
|
|
'''))
|
|
write_to(os.path.join(dirpath, 'e.sls'), textwrap.dedent('''\
|
|
#!pydsl
|
|
success = True
|
|
'''))
|
|
self.state_highstate({'base': ['b']}, dirpath)
|
|
self.state_highstate({'base': ['c', 'd']}, dirpath)
|
|
|
|
|
|
def write_to(fpath, content):
|
|
with salt.utils.files.fopen(fpath, 'w') as f:
|
|
f.write(content)
|