ad6373430b
Pretty straightforward. Give it a URL with an exported GPG key for signing an Apt repository. It downloads it and will install it using apt-key. It's even smart enough to tell if it's already there (i.e. actually tells you if it changed or not).
352 lines
14 KiB
Python
352 lines
14 KiB
Python
|
|
# tests are fairly 'live' (but safe to run)
|
|
# setup authorized_keys for logged in user such
|
|
# that the user can log in as themselves before running tests
|
|
|
|
import unittest
|
|
import getpass
|
|
import ansible.runner
|
|
import os
|
|
import shutil
|
|
import time
|
|
import tempfile
|
|
import urllib2
|
|
|
|
from nose.plugins.skip import SkipTest
|
|
|
|
def get_binary(name):
|
|
for directory in os.environ["PATH"].split(os.pathsep):
|
|
path = os.path.join(directory, name)
|
|
if os.path.isfile(path) and os.access(path, os.X_OK):
|
|
return path
|
|
return None
|
|
|
|
class TestRunner(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.user = getpass.getuser()
|
|
self.runner = ansible.runner.Runner(
|
|
module_name='ping',
|
|
module_path='library/',
|
|
module_args='',
|
|
remote_user=self.user,
|
|
remote_pass=None,
|
|
host_list='test/ansible_hosts',
|
|
timeout=5,
|
|
forks=1,
|
|
background=0,
|
|
pattern='all',
|
|
transport='local',
|
|
)
|
|
self.cwd = os.getcwd()
|
|
self.test_dir = os.path.join(self.cwd, 'test')
|
|
self.stage_dir = self._prepare_stage_dir()
|
|
|
|
def _prepare_stage_dir(self):
|
|
stage_path = os.path.join(self.test_dir, 'test_data')
|
|
if os.path.exists(stage_path):
|
|
shutil.rmtree(stage_path, ignore_errors=False)
|
|
assert not os.path.exists(stage_path)
|
|
os.makedirs(stage_path)
|
|
assert os.path.exists(stage_path)
|
|
return stage_path
|
|
|
|
def _get_test_file(self, filename):
|
|
# get a file inside the test input directory
|
|
filename = os.path.join(self.test_dir, filename)
|
|
assert os.path.exists(filename)
|
|
return filename
|
|
|
|
def _get_stage_file(self, filename):
|
|
# get a file inside the test output directory
|
|
filename = os.path.join(self.stage_dir, filename)
|
|
return filename
|
|
|
|
def _run(self, module_name, module_args, background=0):
|
|
''' run a module and get the localhost results '''
|
|
self.runner.module_name = module_name
|
|
args = ' '.join(module_args)
|
|
print "DEBUG: using args=%s" % args
|
|
self.runner.module_args = args
|
|
self.runner.background = background
|
|
results = self.runner.run()
|
|
# when using nosetests this will only show up on failure
|
|
# which is pretty useful
|
|
print "RESULTS=%s" % results
|
|
assert "localhost" in results['contacted']
|
|
return results['contacted']['localhost']
|
|
|
|
def test_ping(self):
|
|
result = self._run('ping', [])
|
|
assert "ping" in result
|
|
|
|
def test_facter(self):
|
|
if not get_binary("facter"):
|
|
raise SkipTest
|
|
result = self._run('facter', [])
|
|
assert "hostname" in result
|
|
|
|
# temporarily disbabled since it occasionally hangs
|
|
# ohai's fault, setup module doesn't actually run this
|
|
# to get ohai's "facts" anyway
|
|
#
|
|
#def test_ohai(self):
|
|
# if not get_binary("facter"):
|
|
# raise SkipTest
|
|
# result = self._run('ohai',[])
|
|
# assert "hostname" in result
|
|
|
|
def test_copy(self):
|
|
# test copy module, change trigger, etc
|
|
input_ = self._get_test_file('sample.j2')
|
|
output = self._get_stage_file('sample.out')
|
|
assert not os.path.exists(output)
|
|
result = self._run('copy', [
|
|
"src=%s" % input_,
|
|
"dest=%s" % output,
|
|
])
|
|
assert os.path.exists(output)
|
|
data_in = file(input_).read()
|
|
data_out = file(output).read()
|
|
assert data_in == data_out
|
|
assert 'failed' not in result
|
|
assert result['changed'] == True
|
|
assert 'md5sum' in result
|
|
result = self._run('copy', [
|
|
"src=%s" % input_,
|
|
"dest=%s" % output,
|
|
])
|
|
assert result['changed'] == False
|
|
|
|
def test_command(self):
|
|
# test command module, change trigger, etc
|
|
result = self._run('command', [ "/bin/echo", "hi" ])
|
|
assert "failed" not in result
|
|
assert "msg" not in result
|
|
assert result['rc'] == 0
|
|
assert result['stdout'] == 'hi'
|
|
assert result['stderr'] == ''
|
|
|
|
result = self._run('command', [ "false" ])
|
|
assert result['rc'] == 1
|
|
assert 'failed' not in result
|
|
|
|
result = self._run('command', [ "/usr/bin/this_does_not_exist", "splat" ])
|
|
assert 'msg' in result
|
|
assert 'failed' in result
|
|
|
|
result = self._run('shell', [ "/bin/echo", "$HOME" ])
|
|
assert 'failed' not in result
|
|
assert result['rc'] == 0
|
|
|
|
result = self._run('command', [ "creates='/tmp/ansible command test'", "chdir=/tmp", "touch", "'ansible command test'" ])
|
|
assert 'changed' in result
|
|
assert result['rc'] == 0
|
|
|
|
result = self._run('command', [ "creates='/tmp/ansible command test'", "false" ])
|
|
assert 'skipped' in result
|
|
|
|
result = self._run('shell', [ "removes=/tmp/ansible\\ command\\ test", "chdir=/tmp", "rm -f 'ansible command test'; echo $?" ])
|
|
assert 'changed' in result
|
|
assert result['rc'] == 0
|
|
assert result['stdout'] == '0'
|
|
|
|
result = self._run('shell', [ "removes=/tmp/ansible\\ command\\ test", "false" ])
|
|
assert 'skipped' in result
|
|
|
|
def test_git(self):
|
|
self._run('file',['path=/tmp/gitdemo','state=absent'])
|
|
self._run('file',['path=/tmp/gd','state=absent'])
|
|
self._run('command',['git init gitdemo', 'chdir=/tmp'])
|
|
self._run('command',['touch a', 'chdir=/tmp/gitdemo'])
|
|
self._run('command',['git add *', 'chdir=/tmp/gitdemo'])
|
|
self._run('command',['git commit -m "test commit 2"', 'chdir=/tmp/gitdemo'])
|
|
self._run('command',['touch b', 'chdir=/tmp/gitdemo'])
|
|
self._run('command',['git add *', 'chdir=/tmp/gitdemo'])
|
|
self._run('command',['git commit -m "test commit 2"', 'chdir=/tmp/gitdemo'])
|
|
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd" ])
|
|
assert result['changed']
|
|
# test the force option not set
|
|
self._run('file',['path=/tmp/gd/a','state=absent'])
|
|
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd", "force=no" ])
|
|
assert result['failed']
|
|
# test the force option when set
|
|
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd", "force=yes" ])
|
|
assert result['changed']
|
|
|
|
def test_file(self):
|
|
filedemo = tempfile.mkstemp()[1]
|
|
assert self._run('file', ['dest=' + filedemo, 'state=directory'])['failed']
|
|
assert os.path.isfile(filedemo)
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])['failed']
|
|
assert os.path.isfile(filedemo)
|
|
|
|
res = self._run('file', ['dest=' + filedemo, 'mode=604', 'state=file'])
|
|
print res
|
|
assert res['changed']
|
|
assert os.path.isfile(filedemo) and os.stat(filedemo).st_mode == 0100604
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
assert not os.path.exists(filedemo)
|
|
assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
|
|
|
|
filedemo = tempfile.mkdtemp()
|
|
assert self._run('file', ['dest=' + filedemo, 'state=file'])['failed']
|
|
assert os.path.isdir(filedemo)
|
|
|
|
# this used to fail but will now make a 'null' symlink in the directory pointing to dev/null.
|
|
# I feel this is ok but don't want to enforce it with a test.
|
|
#result = self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])
|
|
#assert result['failed']
|
|
#assert os.path.isdir(filedemo)
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'mode=701', 'state=directory'])['changed']
|
|
assert os.path.isdir(filedemo) and os.stat(filedemo).st_mode == 040701
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
assert not os.path.exists(filedemo)
|
|
assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
|
|
|
|
tmp_dir = tempfile.mkdtemp()
|
|
filedemo = os.path.join(tmp_dir, 'link')
|
|
os.symlink('/dev/zero', filedemo)
|
|
assert self._run('file', ['dest=' + filedemo, 'state=file'])['failed']
|
|
assert os.path.islink(filedemo)
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'state=directory'])['failed']
|
|
assert os.path.islink(filedemo)
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])['changed']
|
|
assert os.path.islink(filedemo) and os.path.realpath(filedemo) == '/dev/null'
|
|
|
|
assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
assert not os.path.exists(filedemo)
|
|
assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
|
|
os.rmdir(tmp_dir)
|
|
|
|
def test_large_output(self):
|
|
large_path = "/usr/share/dict/words"
|
|
if not os.path.exists(large_path):
|
|
large_path = "/usr/share/dict/cracklib-small"
|
|
if not os.path.exists(large_path):
|
|
raise SkipTest
|
|
# Ensure reading a large amount of output from a command doesn't hang.
|
|
result = self._run('command', [ "/bin/cat", large_path ])
|
|
assert "failed" not in result
|
|
assert "msg" not in result
|
|
assert result['rc'] == 0
|
|
assert len(result['stdout']) > 100000
|
|
assert result['stderr'] == ''
|
|
|
|
def test_async(self):
|
|
# test async launch and job status
|
|
# of any particular module
|
|
result = self._run('command', [ get_binary("sleep"), "3" ], background=20)
|
|
assert 'ansible_job_id' in result
|
|
assert 'started' in result
|
|
jid = result['ansible_job_id']
|
|
# no real chance of this op taking a while, but whatever
|
|
time.sleep(5)
|
|
# CLI will abstract this (when polling), but this is how it works internally
|
|
result = self._run('async_status', [ "jid=%s" % jid ])
|
|
# TODO: would be nice to have tests for supervisory process
|
|
# killing job after X seconds
|
|
assert 'finished' in result
|
|
assert 'failed' not in result
|
|
assert 'rc' in result
|
|
assert 'stdout' in result
|
|
assert result['ansible_job_id'] == jid
|
|
|
|
def test_fetch(self):
|
|
input_ = self._get_test_file('sample.j2')
|
|
output = os.path.join(self.stage_dir, 'localhost', input_)
|
|
result = self._run('fetch', [ "src=%s" % input_, "dest=%s" % self.stage_dir ])
|
|
assert os.path.exists(output)
|
|
assert open(input_).read() == open(output).read()
|
|
|
|
def test_assemble(self):
|
|
input = self._get_test_file('assemble.d')
|
|
output = self._get_stage_file('sample.out')
|
|
result = self._run('assemble', [
|
|
"src=%s" % input,
|
|
"dest=%s" % output,
|
|
])
|
|
print result
|
|
assert os.path.exists(output)
|
|
out = file(output).read()
|
|
assert out.find("first") != -1
|
|
assert out.find("second") != -1
|
|
assert out.find("third") != -1
|
|
assert result['changed'] == True
|
|
assert 'md5sum' in result
|
|
assert 'failed' not in result
|
|
result = self._run('assemble', [
|
|
"src=%s" % input,
|
|
"dest=%s" % output,
|
|
])
|
|
print result
|
|
assert result['changed'] == False
|
|
|
|
def test_apt_key(self):
|
|
try:
|
|
key_file = self._get_test_file("apt_key.gpg")
|
|
key_file_url = 'file://' + urllib2.quote(key_file)
|
|
key_id = '473041FA'
|
|
|
|
os.environ['ANSIBLE_TEST_APT_KEY'] = 'none'
|
|
# key missing, should download and add
|
|
result = self._run('apt_key', ['state=present', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 1
|
|
assert result['downloaded'] == 1
|
|
assert result['removed'] == 0
|
|
assert result['changed']
|
|
|
|
os.environ["ANSIBLE_TEST_APT_KEY"] = key_id
|
|
# key missing, shouldn't download, no changes
|
|
result = self._run('apt_key', ['id=12345678', 'state=absent', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 0
|
|
assert result['removed'] == 0
|
|
assert not result['changed']
|
|
# key missing, should download and fail sanity check, no changes
|
|
result = self._run('apt_key', ['id=12345678', 'state=present', 'url=' + key_file_url])
|
|
assert 'failed' in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 1
|
|
assert result['removed'] == 0
|
|
# key present, shouldn't download, no changes
|
|
result = self._run('apt_key', ['id=' + key_id, 'state=present', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 0
|
|
assert result['removed'] == 0
|
|
assert not result['changed']
|
|
# key present, should download to get key id
|
|
result = self._run('apt_key', ['state=present', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 1
|
|
assert result['removed'] == 0
|
|
assert not result['changed']
|
|
# key present, should download to get key id and remove
|
|
result = self._run('apt_key', ['state=absent', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 1
|
|
assert result['removed'] == 1
|
|
assert result['changed']
|
|
# key present, should remove but not download
|
|
result = self._run('apt_key', ['id=' + key_id, 'state=absent', 'url=' + key_file_url])
|
|
assert 'failed' not in result
|
|
assert result['added'] == 0
|
|
assert result['downloaded'] == 0
|
|
assert result['removed'] == 1
|
|
assert result['changed']
|
|
finally:
|
|
# always clean up the environment
|
|
os.environ.pop('ANSIBLE_TEST_APT_KEY', None)
|