valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 08:58:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #9152 from cachedout/roots_gitfs_testing

Browse Source 
Fileserver integration testing.
This commit is contained in:
Thomas S Hatch 2013-12-10 16:47:57 -08:00
commit 86e58abe54
7 changed files with 163 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/integration/files/file/base/dest_sym Symbolic link
View File

@ -0,0 +1 @@
source_sym

0
tests/integration/files/file/base/source_sym Normal file
View File

0
tests/integration/files/file/base/test_deep/a/test.sls Normal file
View File

0
tests/integration/files/file/base/test_deep/b/2/test.sls Normal file
View File

0
tests/integration/files/file/base/test_deep/test.sls Normal file
View File

59
tests/integration/fileserver/gitfs_test.py Normal file
View File

@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
'''
:codauthor: :email:`Mike Place <mp@saltstack.com>`
'''
# Import Salt Testing libs
from salttesting import skipIf
from salttesting.helpers import (ensure_in_syspath, destructiveTest)
from salttesting.mock import patch, MagicMock, call, NO_MOCK, NO_MOCK_REASON
ensure_in_syspath('../')
# Import Python libs
import os
import shutil
# Import salt libs
import integration
from salt.fileserver import gitfs
gitfs.__opts__ = {'gitfs_remotes': [''],
'gitfs_root': os.path.join(integration.TMP, 'gitfs_root'),
'fileserver_backend': 'gitfs',
}
GITFS_AVAILABLE = None
try:
import git
GITFS_AVAILABLE = True
except ImportError:
pass
if not gitfs.__virtual__():
GITFS_AVAILABLE = False
@skipIf(not GITFS_AVAILABLE, "GitFS could not be loaded. Skipping GitFS tests!")
class GitFSTest(integration.ModuleCase):
def setUp(self):
'''
We don't want to check in another .git dir into GH because that just gets messy.
Instead, we'll create a temporary repo on the fly for the tests to examine.
:return:
'''
integration_base_files = os.path.join(integration.FILES, 'file', 'base')
tmp_repo_dir = os.path.join(integration.TMP, 'gitfs_root')
tmp_repo_git = os.path.join(tmp_repo_dir, '.git')
# Create the dir if it doesn't already exist
os.mkdirs(tmp_repo_git)
git.Repo.create(tmp_repo_git, bare=True)
shutil.copytree(integration_base_files, tmp_repo_dir, symlinks=True)
git_bin = git.Git(tmp_repo_git)
git_bin.add(tmp_repo_dir)
def test_fake(self):
pass

146
tests/integration/fileserver/roots_test.py
View File

@ -4,24 +4,124 @@
'''
# Import Salt Testing libs
from salttesting import skipIf
from salttesting.helpers import (ensure_in_syspath, destructiveTest)
from salttesting.mock import patch, MagicMock, call, NO_MOCK, NO_MOCK_REASON
ensure_in_syspath('../')
# Import salt libs
import integration
from salt.fileserver import roots
from salt import fileclient
roots.__opts__ = {}
# Import Python libs
import os
import shutil
class RootsTest(integration.ModuleCase):
def setUp(self):
self.master_opts['file_roots']['base'] = [os.path.join(integration.FILES, 'file', 'base')]
def test_file_list(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False}):
ret = roots.file_list({'saltenv': 'base'})
self.assertIn('testfile', ret)
def test_find_file(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False}):
ret = roots.find_file('testfile')
self.assertEqual('testfile', ret['rel'])
full_path_to_file = os.path.join(integration.FILES, 'file', 'base', 'testfile')
self.assertEqual(full_path_to_file, ret['path'])
def test_serve_file(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False,
'file_buffer_size': 262144}):
load = {'saltenv': 'base',
'path': os.path.join(integration.FILES, 'file', 'base', 'testfile'),
'loc': 0
}
fnd = {'path': os.path.join(integration.FILES, 'file', 'base', 'testfile'),
'rel': 'testfile'}
ret = roots.serve_file(load, fnd)
self.assertDictEqual(ret,
{'data': 'Scene 24\n\n \n OLD MAN: Ah, hee he he ha!\n ARTHUR: And this enchanter of whom you speak, he has seen the grail?\n OLD MAN: Ha ha he he he he!\n ARTHUR: Where does he live? Old man, where does he live?\n OLD MAN: He knows of a cave, a cave which no man has entered.\n ARTHUR: And the Grail... The Grail is there?\n OLD MAN: Very much danger, for beyond the cave lies the Gorge\n of Eternal Peril, which no man has ever crossed.\n ARTHUR: But the Grail! Where is the Grail!?\n OLD MAN: Seek you the Bridge of Death.\n ARTHUR: The Bridge of Death, which leads to the Grail?\n OLD MAN: Hee hee ha ha!\n\n',
'dest': 'testfile'})
@skipIf(True, "Update test not yet implemented")
def test_update(self):
pass
def test_file_hash(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False,
'hash_type': self.master_opts['hash_type'],
'cachedir': self.master_opts['cachedir']}):
load = {
'saltenv': 'base',
'path': os.path.join(integration.FILES, 'file', 'base', 'testfile'),
}
fnd = {
'path': os.path.join(integration.FILES, 'file', 'base', 'testfile'),
'rel': 'testfile'
}
ret = roots.file_hash(load, fnd)
self.assertDictEqual(ret, {'hsum': '98aa509006628302ce38ce521a7f805f', 'hash_type': 'md5'})
def test_file_list_emptydirs(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False}):
ret = roots.file_list_emptydirs({'saltenv': 'base'})
self.assertIn('empty_dir', ret)
def test_dir_list(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False}):
ret = roots.dir_list({'saltenv': 'base'})
self.assertIn('empty_dir', ret)
def test_symlink_list(self):
with patch.dict(roots.__opts__, {'file_roots': self.master_opts['file_roots'],
'fileserver_ignoresymlinks': False,
'fileserver_followsymlinks': False,
'file_ignore_regex': False,
'file_ignore_glob': False}):
ret = roots.symlink_list({'saltenv': 'base'})
self.assertDictEqual(ret, {'dest_sym': 'source_sym'})
class RootsLimitTraversalTest(integration.ModuleCase):
@destructiveTest
def setUp(self):
self._create_test_deep_structure()
self.master_opts['file_roots']['base'] = [os.path.join(integration.FILES, 'file', 'base')]
@destructiveTest
# @destructiveTest
def test_limit_traversal(self):
'''
1) Set up a deep directory structure
@ -37,46 +137,6 @@ class RootsLimitTraversalTest(integration.ModuleCase):
self.assertIn('test_deep.a.test', ret)
self.assertNotIn('test_deep.b.2.test', ret)
@destructiveTest
def _create_test_deep_structure(self):
'''
Create a directory structure as follows, comments indicate whether a traversal should find SLS
files above.
test_deep/test.sls # True
test_deep/a/test.sls # True
test_deep/b/ # True
test_deep/b/1/ # False
test_deep/b/2/test.sls # False
'''
# Get file server root
file_root = self.master_opts['file_roots']['base'][0]
if os.path.exists(os.path.join(file_root, 'test_deep')):
return
for test_dir in ['a', 'b']:
os.makedirs(os.path.join(file_root, 'test_deep', test_dir))
# test_deep/test.sls
open(os.path.join(file_root, 'test_deep', 'test.sls'), 'w').close()
# test_deep/a/test.sls
open(os.path.join(file_root, 'test_deep', 'a', 'test.sls'), 'w').close()
# test_deep/b/1/
os.mkdir(os.path.join(file_root, 'test_deep', 'b', '1'))
# test_deep/b/2/
os.mkdir(os.path.join(file_root, 'test_deep', 'b', '2'))
# test_deep/b/2/test.sls
open(os.path.join(file_root, 'test_deep', 'b', '2', 'test.sls'), 'w').close()
@destructiveTest
def tearDown(self):
test_deep_dir = os.path.join(self.master_opts['file_roots']['base'][0], 'test_deep')
if os.path.exists(test_deep_dir):
os.rmdir(test_deep_dir)
if __name__ == '__main__':
from integration import run_tests
run_tests(RootsLimitTraversalTest)