community.general/test/TestRunner.py

# 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)