diff --git a/test/TestRunner.py b/test/TestRunner.py index 0734b9e3015..9f844ee44b7 100644 --- a/test/TestRunner.py +++ b/test/TestRunner.py @@ -87,130 +87,6 @@ class TestRunner(unittest.TestCase): 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_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 result.get('failed') - - 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 result.get('changed') - assert result['rc'] == 0 - - result = self._run('command', ["creates='/tmp/ansible command test'", "false"]) - assert result.get('skipped') - - result = self._run('shell', ["removes=/tmp/ansible\\ command\\ test", "chdir=/tmp", "rm -f 'ansible command test'; echo $?"]) - assert result.get('changed') - assert result['rc'] == 0 - assert result['stdout'] == '0' - - result = self._run('shell', ["removes=/tmp/ansible\\ command\\ test", "false"]) - assert result.get('skipped') - - def test_git(self): - pid = os.getpid() - git_demo = "gitdemo%d" % pid - git_dm = "gitdm%d" % pid - git_bare = "gdbare%d" % pid - git_ref = "gdref%d" % pid - git_reftest = "gdreftest%d" % pid - self._run('file', ['path=/tmp/%s' % git_demo, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_dm, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_bare, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_ref, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_reftest, 'state=absent']) - self._run('command', ['git init %s' % git_demo, 'chdir=/tmp']) - self._run('command', ['touch a', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git add *', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git commit -m "test commit 1"', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['touch b', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git add *', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git commit -m "test commit 2"', 'chdir=/tmp/%s' % git_demo]) - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_dm]) - assert result['changed'] - # test the force option not set - self._run('file', ['path=/tmp/%s/a' % git_dm, 'state=absent']) - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_dm, "force=no"]) - assert result['failed'] - # test the force option when set - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_dm, "force=yes"]) - assert result['changed'] - # test the bare option - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_bare, "bare=yes", "remote=test"]) - assert result['changed'] - # test a no-op fetch, add origin for el6 versions of git - self._run('command', ['git', 'remote', 'add', 'origin', 'file:///tmp/%s' % git_demo, 'chdir=/tmp/%s' % git_bare]) - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_bare, "bare=yes"]) - assert not result['changed'] - # test whether fetch is working for bare repos - self._run('command', ['touch c', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git add *', 'chdir=/tmp/%s' % git_demo]) - self._run('command', ['git commit -m "test commit 3"', 'chdir=/tmp/%s' % git_demo]) - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_bare, "bare=yes"]) - assert result['changed'] - # test reference repos - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_bare, "dest=/tmp/%s" % git_ref, "bare=yes"]) - assert result['changed'] - result = self._run('git', ["repo=\"file:///tmp/%s\"" % git_demo, "dest=/tmp/%s" % git_reftest, "reference=/tmp/%s/" % git_ref]) - assert result['changed'] - assert os.path.isfile('/tmp/%s/a' % git_reftest) - result = self._run('command', ['ls', 'chdir=/tmp/%s/objects/pack' % git_ref]) - assert result['stdout'] != '' - result = self._run('command', ['ls', 'chdir=/tmp/%s/.git/objects/pack' % git_reftest]) - assert result['stdout'] == '' - # cleanup - self._run('file', ['path=/tmp/%s' % git_demo, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_dm, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_bare, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_ref, 'state=absent']) - self._run('file', ['path=/tmp/%s' % git_reftest, 'state=absent']) - - 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 @@ -251,339 +127,3 @@ class TestRunner(unittest.TestCase): ]) assert result['changed'] is False - def test_lineinfile(self): - # Unit tests for the lineinfile module, without backref features. - sampleroot = 'rocannon' - sample_origin = self._get_test_file(sampleroot + '.txt') - sample = self._get_stage_file(sampleroot + '.out' + '.txt') - shutil.copy(sample_origin, sample) - # The order of the test cases is important - - # defaults to insertafter at the end of the file - testline = 'First: Line added by default at the end of the file.' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^First: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact[-1] == testline - assert artifact.count(testline) == 1 - - # run a second time, verify only one line has been added - result = self._run(*testcase) - assert not result['changed'] - assert result['msg'] == '' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline) == 1 - - # insertafter with EOF - testline = 'Second: Line added with insertafter=EOF' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter=EOF", - "regexp='^Second: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact[-1] == testline - assert artifact.count(testline) == 1 - - # with invalid insertafter regex - # If the regexp doesn't match and the insertafter doesn't match, - # do nothing. - testline = 'Third: Line added with an invalid insertafter regex' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='^abcdefgh'", - "regexp='^Third: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert not result['changed'] - - # with an insertafter regex - # The regexp doesn't match, but the insertafter is specified and does, - # so insert after insertafter. - testline = 'Fourth: Line added with a valid insertafter regex' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='^receive messages to '", - "regexp='^Fourth: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline) == 1 - idx = artifact.index('receive messages to and from a corresponding device over any distance') - assert artifact[idx + 1] == testline - - # replacement of a line from a regex - # we replace the line, so we need to get its idx before the run - artifact = [x.strip() for x in open(sample)] - target_line = 'combination of microphone, speaker, keyboard and display. It can send and' - idx = artifact.index(target_line) - - testline = 'Fith: replacement of a line: combination of microphone' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='combination of microphone'", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line replaced' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline) == 1 - assert artifact.index(testline) == idx - assert target_line not in artifact - - # removal of a line - # we replace the line, so we need to get its idx before the run - artifact = [x.strip() for x in open(sample)] - target_line = 'receive messages to and from a corresponding device over any distance' - idx = artifact.index(target_line) - - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^receive messages to and from '", - "state=absent" - ]) - result = self._run(*testcase) - assert result['changed'] - artifact = [x.strip() for x in open(sample)] - assert target_line not in artifact - - # with both insertafter and insertbefore (should fail) - testline = 'Seventh: this line should not be there' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='BOF'", - "insertbefore='BOF'", - "regexp='^communication. '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['failed'] - - # insertbefore with BOF - testline = 'Eighth: insertbefore BOF' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertbefore=BOF", - "regexp='^Eighth: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline) == 1 - assert artifact[0] == testline - - # insertbefore with regex - testline = 'Ninth: insertbefore with a regex' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertbefore='^communication. Typically '", - "regexp='^Ninth: '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline) == 1 - idx = artifact.index('communication. Typically it is depicted as a lunch-box sized object with some') - assert artifact[idx - 1] == testline - - # Testing validate - testline = 'Tenth: Testing with validate' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^Tenth: '", - "line='%s'" % testline, - "validate='grep -q Tenth %s'", - ]) - result = self._run(*testcase) - assert result['changed'], "File wasn't changed when it should have been" - assert result['msg'] == 'line added', "msg was incorrect" - artifact = [ x.strip() for x in open(sample) ] - assert artifact[-1] == testline - - - # Testing validate - testline = '#11: Testing with validate' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^#11: '", - "line='%s'" % testline, - "validate='grep -q #12# %s'", - ]) - result = self._run(*testcase) - assert result['failed'] - - # insert multiline at the end of the file - testline1 = '#12: The \\n character replaced with' - testline2 = 'an actual newline.' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^#12: '", - "line='%s\n%s'" % (testline1, testline2) - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line added' - artifact = [x.strip() for x in open(sample)] - assert artifact[-2] == testline1 - assert artifact[-1] == testline2 - assert artifact.count(testline1) == 1 - assert artifact.count(testline2) == 1 - - # cleanup - os.unlink(sample) - - def test_lineinfile_backrefs(self): - # Unit tests for the lineinfile module, with backref features. - sampleroot = 'rocannon' - sample_origin = self._get_test_file(sampleroot + '.txt') - origin_lines = [line.strip() for line in open(sample_origin)] - sample = self._get_stage_file(sampleroot + '.out' + '.txt') - shutil.copy(sample_origin, sample) - # The order of the test cases is important - - # The regexp doesn't match, so the line will not be added anywhere. - testline = r'\1: Line added by default at the end of the file.' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='^(First): '", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert not result['changed'] - assert result['msg'] == '' - artifact = [x.strip() for x in open(sample)] - assert artifact == origin_lines - - # insertafter with EOF - # The regexp doesn't match, so the line will not be added anywhere. - testline = r'\1: Line added with insertafter=EOF' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter=EOF", - "regexp='^(Second): '", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert not result['changed'] - assert result['msg'] == '' - artifact = [x.strip() for x in open(sample)] - assert artifact == origin_lines - - # with invalid insertafter regex - # The regexp doesn't match, so do nothing. - testline = r'\1: Line added with an invalid insertafter regex' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='^abcdefgh'", - "regexp='^(Third): '", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert not result['changed'] - assert artifact == origin_lines - - # with an insertafter regex - # The regexp doesn't match, so do nothing. - testline = r'\1: Line added with a valid insertafter regex' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='^receive messages to '", - "regexp='^(Fourth): '", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert not result['changed'] - assert result['msg'] == '' - assert artifact == origin_lines - - # replacement of a line from a regex - # we replace the line, so we need to get its idx before the run - artifact = [x.strip() for x in open(sample)] - target_line = 'combination of microphone, speaker, keyboard and display. It can send and' - idx = artifact.index(target_line) - - testline = r'\1 of megaphone' - testline_after = 'combination of megaphone' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='(combination) of microphone'", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert result['changed'] - assert result['msg'] == 'line replaced' - artifact = [x.strip() for x in open(sample)] - assert artifact.count(testline_after) == 1 - assert artifact.index(testline_after) == idx - assert target_line not in artifact - - # Go again, should be unchanged now. - testline = r'\1 of megaphone' - testline_after = 'combination of megaphone' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "regexp='(combination) of megaphone'", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - assert not result['changed'] - assert result['msg'] == '' - - # Try a numeric, named capture group example. - f = open(sample, 'a+') - f.write("1 + 1 = 3" + os.linesep) - f.close() - testline = r"2 + \g = 3" - testline_after = "2 + 1 = 3" - testcase = ('lineinfile', [ - "dest=%s" % sample, - r"regexp='1 \+ (?P\d) = 3'", - "line='%s'" % testline, - "backrefs=yes", - ]) - result = self._run(*testcase) - artifact = [x.strip() for x in open(sample)] - assert result['changed'] - assert result['msg'] == 'line replaced' - artifact = [x.strip() for x in open(sample)] - assert '1 + 1 = 3' not in artifact - assert testline_after == artifact[-1] - - # with both insertafter and insertbefore (should fail) - testline = 'Seventh: this line should not be there' - testcase = ('lineinfile', [ - "dest=%s" % sample, - "insertafter='BOF'", - "insertbefore='BOF'", - "regexp='^communication. '", - "line='%s'" % testline - ]) - result = self._run(*testcase) - assert result['failed'] - - os.unlink(sample)