mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
33a7f8b2ec
lintian found several spelling errors. Signed-off-by: Benjamin Drung <benjamin.drung@profitbricks.com>
610 lines
22 KiB
Python
610 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
'''
|
|
Tests for the Git state
|
|
'''
|
|
|
|
# Import python libs
|
|
from __future__ import absolute_import
|
|
import errno
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import string
|
|
import subprocess
|
|
import tempfile
|
|
|
|
# Import Salt Testing libs
|
|
from salttesting.helpers import ensure_in_syspath, skip_if_binaries_missing
|
|
ensure_in_syspath('../../')
|
|
|
|
# Import salt libs
|
|
import integration
|
|
import salt.utils
|
|
|
|
|
|
@skip_if_binaries_missing('git')
|
|
class GitTest(integration.ModuleCase, integration.SaltReturnAssertsMixIn):
|
|
'''
|
|
Validate the git state
|
|
'''
|
|
|
|
def setUp(self):
|
|
super(GitTest, self).setUp()
|
|
self.__domain = 'github.com'
|
|
try:
|
|
if hasattr(socket, 'setdefaulttimeout'):
|
|
# 10 second dns timeout
|
|
socket.setdefaulttimeout(10)
|
|
socket.gethostbyname(self.__domain)
|
|
except socket.error:
|
|
msg = 'error resolving {0}, possible network issue?'
|
|
self.skipTest(msg.format(self.__domain))
|
|
|
|
def tearDown(self):
|
|
# Reset the dns timeout after the test is over
|
|
socket.setdefaulttimeout(None)
|
|
|
|
def test_latest(self):
|
|
'''
|
|
git.latest
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
target=name
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_with_rev_and_submodules(self):
|
|
'''
|
|
git.latest
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev='develop',
|
|
target=name,
|
|
submodules=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_failure(self):
|
|
'''
|
|
git.latest
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://youSpelledGitHubWrong.com/saltstack/salt-test-repo.git',
|
|
rev='develop',
|
|
target=name,
|
|
submodules=True
|
|
)
|
|
self.assertSaltFalseReturn(ret)
|
|
self.assertFalse(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_empty_dir(self):
|
|
'''
|
|
git.latest
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
if not os.path.isdir(name):
|
|
os.mkdir(name)
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev='develop',
|
|
target=name,
|
|
submodules=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_unless_no_cwd_issue_6800(self):
|
|
'''
|
|
cwd=target was being passed to _run_check which blew up if
|
|
target dir did not already exist.
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
if os.path.isdir(name):
|
|
shutil.rmtree(name)
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev='develop',
|
|
target=name,
|
|
unless='test -e {0}'.format(name),
|
|
submodules=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_numeric_rev(self):
|
|
'''
|
|
git.latest with numeric revision
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev=0.11,
|
|
target=name,
|
|
submodules=True,
|
|
timeout=120
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_with_local_changes(self):
|
|
'''
|
|
Ensure that we fail the state when there are local changes and succeed
|
|
when force_reset is True.
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
# Clone repo
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
target=name
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isdir(os.path.join(name, '.git')))
|
|
|
|
# Make change to LICENSE file.
|
|
with salt.utils.fopen(os.path.join(name, 'LICENSE'), 'a') as fp_:
|
|
fp_.write('Lorem ipsum dolor blah blah blah....\n')
|
|
|
|
# Make sure that we now have uncommitted changes
|
|
self.assertTrue(self.run_function('git.diff', [name, 'HEAD']))
|
|
|
|
# Re-run state with force_reset=False
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
target=name,
|
|
force_reset=False
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertEqual(
|
|
ret[next(iter(ret))]['comment'],
|
|
('Repository {0} is up-to-date, but with local changes. Set '
|
|
'\'force_reset\' to True to purge local changes.'.format(name))
|
|
)
|
|
|
|
# Now run the state with force_reset=True
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
target=name,
|
|
force_reset=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
# Make sure that we no longer have uncommitted changes
|
|
self.assertFalse(self.run_function('git.diff', [name, 'HEAD']))
|
|
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_fast_forward(self):
|
|
'''
|
|
Test running git.latest state a second time after changes have been
|
|
made to the remote repo.
|
|
'''
|
|
def _head(cwd):
|
|
return self.run_function('git.rev_parse', [cwd, 'HEAD'])
|
|
|
|
repo_url = 'https://{0}/saltstack/salt-test-repo.git'.format(self.__domain)
|
|
mirror_dir = os.path.join(integration.TMP, 'salt_repo_mirror')
|
|
mirror_url = 'file://' + mirror_dir
|
|
admin_dir = os.path.join(integration.TMP, 'salt_repo_admin')
|
|
clone_dir = os.path.join(integration.TMP, 'salt_repo')
|
|
|
|
try:
|
|
# Mirror the repo
|
|
self.run_function('git.clone',
|
|
[mirror_dir, repo_url, None, '--mirror'])
|
|
|
|
# Make sure the directory for the mirror now exists
|
|
self.assertTrue(os.path.exists(mirror_dir))
|
|
|
|
# Clone the mirror twice, once to the admin location and once to
|
|
# the clone_dir
|
|
ret = self.run_state('git.latest', name=mirror_url, target=admin_dir)
|
|
self.assertSaltTrueReturn(ret)
|
|
ret = self.run_state('git.latest', name=mirror_url, target=clone_dir)
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
# Make a change to the repo by editing the file in the admin copy
|
|
# of the repo and committing.
|
|
head_pre = _head(admin_dir)
|
|
with open(os.path.join(admin_dir, 'LICENSE'), 'a') as fp_:
|
|
fp_.write('Hello world!')
|
|
self.run_function('git.commit', [admin_dir, 'Added a line', '-a'])
|
|
# Make sure HEAD is pointing to a new SHA so we know we properly
|
|
# committed our change.
|
|
head_post = _head(admin_dir)
|
|
self.assertNotEqual(head_pre, head_post)
|
|
|
|
# Push the change to the mirror
|
|
# NOTE: the test will fail if the salt-test-repo's default branch
|
|
# is changed.
|
|
self.run_function('git.push', [admin_dir, 'origin', 'develop'])
|
|
|
|
# Re-run the git.latest state on the clone_dir
|
|
ret = self.run_state('git.latest', name=mirror_url, target=clone_dir)
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
# Make sure that the clone_dir now has the correct SHA
|
|
self.assertEqual(head_post, _head(clone_dir))
|
|
|
|
finally:
|
|
for path in (mirror_dir, admin_dir, clone_dir):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
|
|
def _changed_local_branch_helper(self, rev, hint):
|
|
'''
|
|
We're testing two almost identical cases, the only thing that differs
|
|
is the rev used for the git.latest state.
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
cwd = os.getcwd()
|
|
try:
|
|
# Clone repo
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev=rev,
|
|
target=name
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
# Check out a new branch in the clone and make a commit, to ensure
|
|
# that when we re-run the state, it is not a fast-forward change
|
|
os.chdir(name)
|
|
with salt.utils.fopen(os.devnull, 'w') as devnull:
|
|
subprocess.check_call(['git', 'checkout', '-b', 'new_branch'],
|
|
stdout=devnull, stderr=devnull)
|
|
with salt.utils.fopen('foo', 'w'):
|
|
pass
|
|
subprocess.check_call(['git', 'add', '.'],
|
|
stdout=devnull, stderr=devnull)
|
|
subprocess.check_call(['git', 'commit', '-m', 'add file'],
|
|
stdout=devnull, stderr=devnull)
|
|
os.chdir(cwd)
|
|
|
|
# Re-run the state, this should fail with a specific hint in the
|
|
# comment field.
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name='https://{0}/saltstack/salt-test-repo.git'.format(self.__domain),
|
|
rev=rev,
|
|
target=name
|
|
)
|
|
self.assertSaltFalseReturn(ret)
|
|
|
|
comment = ret[next(iter(ret))]['comment']
|
|
self.assertTrue(hint in comment)
|
|
finally:
|
|
# Make sure that we change back to the original cwd even if there
|
|
# was a traceback in the test.
|
|
os.chdir(cwd)
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_latest_changed_local_branch_rev_head(self):
|
|
'''
|
|
Test for presence of hint in failure message when the local branch has
|
|
been changed and a the rev is set to HEAD
|
|
|
|
This test will fail if the default branch for the salt-test-repo is
|
|
ever changed.
|
|
'''
|
|
self._changed_local_branch_helper(
|
|
'HEAD',
|
|
'The default remote branch (develop) differs from the local '
|
|
'branch (new_branch)'
|
|
)
|
|
|
|
def test_latest_changed_local_branch_rev_develop(self):
|
|
'''
|
|
Test for presence of hint in failure message when the local branch has
|
|
been changed and a non-HEAD rev is specified
|
|
'''
|
|
self._changed_local_branch_helper(
|
|
'develop',
|
|
'The desired rev (develop) differs from the name of the local '
|
|
'branch (new_branch)'
|
|
)
|
|
|
|
def test_latest_updated_remote_rev(self):
|
|
'''
|
|
Ensure that we don't exit early when checking for a fast-forward
|
|
'''
|
|
orig_cwd = os.getcwd()
|
|
name = tempfile.mkdtemp(dir=integration.TMP)
|
|
target = os.path.join(integration.TMP, 'test_latest_updated_remote_rev')
|
|
|
|
# Initialize a new git repository
|
|
subprocess.check_call(['git', 'init', '--quiet', name])
|
|
|
|
try:
|
|
os.chdir(name)
|
|
# Set user.name and user.email config attributes if not present
|
|
for key, value in (('user.name', 'Jenkins'),
|
|
('user.email', 'qa@saltstack.com')):
|
|
# Check if key is missing
|
|
keycheck = subprocess.Popen(
|
|
['git', 'config', '--get', key],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
if keycheck.wait() != 0:
|
|
# Set the key if it is not present
|
|
subprocess.check_call(
|
|
['git', 'config', key, value])
|
|
|
|
# Add and commit a file
|
|
with salt.utils.fopen('foo.txt', 'w') as fp_:
|
|
fp_.write('Hello world\n')
|
|
subprocess.check_call(['git', 'add', '.'])
|
|
subprocess.check_call(['git', 'commit', '-qm', 'init'])
|
|
|
|
# Run the state to clone the repo we just created
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name=name,
|
|
target=target,
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
# Add another commit
|
|
with salt.utils.fopen('foo.txt', 'w') as fp_:
|
|
fp_.write('Added a line\n')
|
|
subprocess.check_call(['git', 'commit', '-aqm', 'added a line'])
|
|
|
|
# Run the state again. It should pass, if it doesn't then there was
|
|
# a problem checking whether or not the change is a fast-forward.
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name=name,
|
|
target=target,
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
finally:
|
|
os.chdir(orig_cwd)
|
|
for path in (name, target):
|
|
try:
|
|
shutil.rmtree(path)
|
|
except OSError as exc:
|
|
if exc.errno != errno.ENOENT:
|
|
raise exc
|
|
|
|
def test_present(self):
|
|
'''
|
|
git.present
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
try:
|
|
ret = self.run_state(
|
|
'git.present',
|
|
name=name,
|
|
bare=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isfile(os.path.join(name, 'HEAD')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_present_failure(self):
|
|
'''
|
|
git.present
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
if not os.path.isdir(name):
|
|
os.mkdir(name)
|
|
try:
|
|
fname = os.path.join(name, 'stoptheprocess')
|
|
|
|
with salt.utils.fopen(fname, 'a') as fh_:
|
|
fh_.write('')
|
|
|
|
ret = self.run_state(
|
|
'git.present',
|
|
name=name,
|
|
bare=True
|
|
)
|
|
self.assertSaltFalseReturn(ret)
|
|
self.assertFalse(os.path.isfile(os.path.join(name, 'HEAD')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_present_empty_dir(self):
|
|
'''
|
|
git.present
|
|
'''
|
|
name = os.path.join(integration.TMP, 'salt_repo')
|
|
if not os.path.isdir(name):
|
|
os.mkdir(name)
|
|
try:
|
|
ret = self.run_state(
|
|
'git.present',
|
|
name=name,
|
|
bare=True
|
|
)
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertTrue(os.path.isfile(os.path.join(name, 'HEAD')))
|
|
finally:
|
|
shutil.rmtree(name, ignore_errors=True)
|
|
|
|
def test_config_set_value_with_space_character(self):
|
|
'''
|
|
git.config
|
|
'''
|
|
name = tempfile.mkdtemp(dir=integration.TMP)
|
|
self.addCleanup(shutil.rmtree, name, ignore_errors=True)
|
|
subprocess.check_call(['git', 'init', '--quiet', name])
|
|
|
|
ret = self.run_state(
|
|
'git.config_set',
|
|
name='user.name',
|
|
value='foo bar',
|
|
repo=name,
|
|
**{'global': False})
|
|
self.assertSaltTrueReturn(ret)
|
|
|
|
|
|
@skip_if_binaries_missing('git')
|
|
class LocalRepoGitTest(integration.ModuleCase, integration.SaltReturnAssertsMixIn):
|
|
'''
|
|
Tests which do no require connectivity to github.com
|
|
'''
|
|
def test_renamed_default_branch(self):
|
|
'''
|
|
Test the case where the remote branch has been removed
|
|
https://github.com/saltstack/salt/issues/36242
|
|
'''
|
|
cwd = os.getcwd()
|
|
repo = tempfile.mkdtemp(dir=integration.TMP)
|
|
admin = tempfile.mkdtemp(dir=integration.TMP)
|
|
name = tempfile.mkdtemp(dir=integration.TMP)
|
|
for dirname in (repo, admin, name):
|
|
self.addCleanup(shutil.rmtree, dirname, ignore_errors=True)
|
|
self.addCleanup(os.chdir, cwd)
|
|
|
|
with salt.utils.fopen(os.devnull, 'w') as devnull:
|
|
# Create bare repo
|
|
subprocess.check_call(['git', 'init', '--bare', repo],
|
|
stdout=devnull, stderr=devnull)
|
|
# Clone bare repo
|
|
subprocess.check_call(['git', 'clone', repo, admin],
|
|
stdout=devnull, stderr=devnull)
|
|
|
|
# Create, add, commit, and push file
|
|
os.chdir(admin)
|
|
with salt.utils.fopen('foo', 'w'):
|
|
pass
|
|
subprocess.check_call(['git', 'add', '.'],
|
|
stdout=devnull, stderr=devnull)
|
|
subprocess.check_call(['git', 'commit', '-m', 'init'],
|
|
stdout=devnull, stderr=devnull)
|
|
subprocess.check_call(['git', 'push', 'origin', 'master'],
|
|
stdout=devnull, stderr=devnull)
|
|
|
|
# Change back to the original cwd
|
|
os.chdir(cwd)
|
|
|
|
# Rename remote 'master' branch to 'develop'
|
|
os.rename(
|
|
os.path.join(repo, 'refs', 'heads', 'master'),
|
|
os.path.join(repo, 'refs', 'heads', 'develop')
|
|
)
|
|
|
|
# Run git.latest state. This should successfully clone and fail with a
|
|
# specific error in the comment field.
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name=repo,
|
|
target=name,
|
|
rev='develop',
|
|
)
|
|
self.assertSaltFalseReturn(ret)
|
|
self.assertEqual(
|
|
ret[next(iter(ret))]['comment'],
|
|
'Remote HEAD refers to a ref that does not exist. '
|
|
'This can happen when the default branch on the '
|
|
'remote repository is renamed or deleted. If you '
|
|
'are unable to fix the remote repository, you can '
|
|
'work around this by setting the \'branch\' argument '
|
|
'(which will ensure that the named branch is created '
|
|
'if it does not already exist).\n\n'
|
|
'Changes already made: {0} cloned to {1}'
|
|
.format(repo, name)
|
|
)
|
|
self.assertEqual(
|
|
ret[next(iter(ret))]['changes'],
|
|
{'new': '{0} => {1}'.format(repo, name)}
|
|
)
|
|
|
|
# Run git.latest state again. This should fail again, with a different
|
|
# error in the comment field, and should not change anything.
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name=repo,
|
|
target=name,
|
|
rev='develop',
|
|
)
|
|
self.assertSaltFalseReturn(ret)
|
|
self.assertEqual(
|
|
ret[next(iter(ret))]['comment'],
|
|
'Cannot set/unset upstream tracking branch, local '
|
|
'HEAD refers to nonexistent branch. This may have '
|
|
'been caused by cloning a remote repository for which '
|
|
'the default branch was renamed or deleted. If you '
|
|
'are unable to fix the remote repository, you can '
|
|
'work around this by setting the \'branch\' argument '
|
|
'(which will ensure that the named branch is created '
|
|
'if it does not already exist).'
|
|
)
|
|
self.assertEqual(ret[next(iter(ret))]['changes'], {})
|
|
|
|
# Run git.latest state again with a branch manually set. This should
|
|
# checkout a new branch and the state should pass.
|
|
ret = self.run_state(
|
|
'git.latest',
|
|
name=repo,
|
|
target=name,
|
|
rev='develop',
|
|
branch='develop',
|
|
)
|
|
# State should succeed
|
|
self.assertSaltTrueReturn(ret)
|
|
self.assertSaltCommentRegexpMatches(
|
|
ret,
|
|
'New branch \'develop\' was checked out, with origin/develop '
|
|
r'\([0-9a-f]{7}\) as a starting point'
|
|
)
|
|
# Only the revision should be in the changes dict.
|
|
self.assertEqual(
|
|
list(ret[next(iter(ret))]['changes'].keys()),
|
|
['revision']
|
|
)
|
|
# Since the remote repo was incorrectly set up, the local head should
|
|
# not exist (therefore the old revision should be None).
|
|
self.assertEqual(
|
|
ret[next(iter(ret))]['changes']['revision']['old'],
|
|
None
|
|
)
|
|
# Make sure the new revision is a SHA (40 chars, all hex)
|
|
self.assertTrue(
|
|
len(ret[next(iter(ret))]['changes']['revision']['new']) == 40)
|
|
self.assertTrue(
|
|
all([x in string.hexdigits for x in
|
|
ret[next(iter(ret))]['changes']['revision']['new']])
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from integration import run_tests
|
|
run_tests(GitTest)
|