forked from saltstack/salt
/
roots_test.py
182 lines (155 loc) · 8.93 KB
/
roots_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Mike Place <mp@saltstack.com>`
'''
# Import Python libs
from __future__ import absolute_import
import os
# Import Salt Testing libs
from salttesting import skipIf
from salttesting.helpers import ensure_in_syspath
from salttesting.mock import patch, NO_MOCK, NO_MOCK_REASON
ensure_in_syspath('../..')
# Import salt libs
import integration
from salt.fileserver import roots
from salt import fileclient
roots.__opts__ = {}
@skipIf(NO_MOCK, NO_MOCK_REASON)
class RootsTest(integration.ModuleCase):
def setUp(self):
if integration.TMP_STATE_TREE not in self.master_opts['file_roots']['base']:
# We need to setup the file roots
self.master_opts['file_roots']['base'] = [os.path.join(integration.FILES, 'file', 'base')]
def test_file_list(self):
with patch.dict(roots.__opts__, {'cachedir': self.master_opts['cachedir'],
'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):
if integration.TMP_STATE_TREE not in self.master_opts['file_roots']['base']:
self.skipTest('This test fails when using tests/runtests.py. salt-runtests will be available soon.')
empty_dir = os.path.join(integration.TMP_STATE_TREE, 'empty_dir')
if not os.path.isdir(empty_dir):
# There's no use creating the empty-directory ourselves at this
# point, the minions have already synced, it wouldn't get pushed to
# them
self.skipTest('This test fails when using tests/runtests.py. salt-runtests will be available soon.')
with patch.dict(roots.__opts__, {'cachedir': self.master_opts['cachedir'],
'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):
empty_dir = os.path.join(integration.TMP_STATE_TREE, 'empty_dir')
if integration.TMP_STATE_TREE not in self.master_opts['file_roots']['base']:
self.skipTest('This test fails when using tests/runtests.py. salt-runtests will be available soon.')
empty_dir = os.path.join(integration.TMP_STATE_TREE, 'empty_dir')
if not os.path.isdir(empty_dir):
# There's no use creating the empty-directory ourselves at this
# point, the minions have already synced, it wouldn't get pushed to
# them
self.skipTest('This test fails when using tests/runtests.py. salt-runtests will be available soon.')
with patch.dict(roots.__opts__, {'cachedir': self.master_opts['cachedir'],
'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__, {'cachedir': self.master_opts['cachedir'],
'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 test_limit_traversal(self):
'''
1) Set up a deep directory structure
2) Enable the configuration option for 'limit_directory_traversal'
3) Ensure that we can find SLS files in a directory so long as there is an SLS file in a directory above.
4) Ensure that we cannot find an SLS file in a directory that does not have an SLS file in a directory above.
'''
file_client_opts = self.get_config('master', from_scratch=True)
file_client_opts['fileserver_limit_traversal'] = True
ret = fileclient.Client(file_client_opts).list_states('base')
self.assertIn('test_deep.test', ret)
self.assertIn('test_deep.a.test', ret)
self.assertNotIn('test_deep.b.2.test', ret)
if __name__ == '__main__':
from integration import run_tests
run_tests(RootsTest, RootsLimitTraversalTest)