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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
# Copyright 2013-2018 Lawrence Livermore National Security, LLC and other
# Spack Project Developers. See the top-level COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
"""Tests for ``llnl/util/filesystem.py``"""
import llnl.util.filesystem as fs
import os
import stat
import pytest
@pytest.fixture()
def stage(tmpdir_factory):
"""Creates a stage with the directory structure for the tests."""
s = tmpdir_factory.mktemp('filesystem_test')
with s.as_cwd():
# Create source file hierarchy
fs.touchp('source/1')
fs.touchp('source/a/b/2')
fs.touchp('source/a/b/3')
fs.touchp('source/c/4')
fs.touchp('source/c/d/5')
fs.touchp('source/c/d/6')
fs.touchp('source/c/d/e/7')
# Create symlinks
os.symlink(os.path.abspath('source/1'), 'source/2')
os.symlink('b/2', 'source/a/b2')
os.symlink('a/b', 'source/f')
# Create destination directory
fs.mkdirp('dest')
yield s
class TestCopy:
"""Tests for ``filesystem.copy``"""
def test_file_dest(self, stage):
"""Test using a filename as the destination."""
with fs.working_dir(str(stage)):
fs.copy('source/1', 'dest/1')
assert os.path.exists('dest/1')
def test_dir_dest(self, stage):
"""Test using a directory as the destination."""
with fs.working_dir(str(stage)):
fs.copy('source/1', 'dest')
assert os.path.exists('dest/1')
def check_added_exe_permissions(src, dst):
src_mode = os.stat(src).st_mode
dst_mode = os.stat(dst).st_mode
for perm in [stat.S_IXUSR, stat.S_IXGRP, stat.S_IXOTH]:
if src_mode & perm:
assert dst_mode & perm
class TestInstall:
"""Tests for ``filesystem.install``"""
def test_file_dest(self, stage):
"""Test using a filename as the destination."""
with fs.working_dir(str(stage)):
fs.install('source/1', 'dest/1')
assert os.path.exists('dest/1')
check_added_exe_permissions('source/1', 'dest/1')
def test_dir_dest(self, stage):
"""Test using a directory as the destination."""
with fs.working_dir(str(stage)):
fs.install('source/1', 'dest')
assert os.path.exists('dest/1')
check_added_exe_permissions('source/1', 'dest/1')
class TestCopyTree:
"""Tests for ``filesystem.copy_tree``"""
def test_existing_dir(self, stage):
"""Test copying to an existing directory."""
with fs.working_dir(str(stage)):
fs.copy_tree('source', 'dest')
assert os.path.exists('dest/a/b/2')
def test_non_existing_dir(self, stage):
"""Test copying to a non-existing directory."""
with fs.working_dir(str(stage)):
fs.copy_tree('source', 'dest/sub/directory')
assert os.path.exists('dest/sub/directory/a/b/2')
def test_symlinks_true(self, stage):
"""Test copying with symlink preservation."""
with fs.working_dir(str(stage)):
fs.copy_tree('source', 'dest', symlinks=True)
assert os.path.exists('dest/2')
assert os.path.islink('dest/2')
assert os.path.exists('dest/a/b2')
with fs.working_dir('dest/a'):
assert os.path.exists(os.readlink('b2'))
assert (os.path.realpath('dest/f/2') ==
os.path.abspath('dest/a/b/2'))
assert os.path.realpath('dest/2') == os.path.abspath('dest/1')
def test_symlinks_true_ignore(self, stage):
"""Test copying when specifying relative paths that should be ignored
"""
with fs.working_dir(str(stage)):
ignore = lambda p: p in ['c/d/e', 'a']
fs.copy_tree('source', 'dest', symlinks=True, ignore=ignore)
assert not os.path.exists('dest/a')
assert os.path.exists('dest/c/d')
assert not os.path.exists('dest/c/d/e')
def test_symlinks_false(self, stage):
"""Test copying without symlink preservation."""
with fs.working_dir(str(stage)):
fs.copy_tree('source', 'dest', symlinks=False)
assert os.path.exists('dest/2')
assert not os.path.islink('dest/2')
class TestInstallTree:
"""Tests for ``filesystem.install_tree``"""
def test_existing_dir(self, stage):
"""Test installing to an existing directory."""
with fs.working_dir(str(stage)):
fs.install_tree('source', 'dest')
assert os.path.exists('dest/a/b/2')
def test_non_existing_dir(self, stage):
"""Test installing to a non-existing directory."""
with fs.working_dir(str(stage)):
fs.install_tree('source', 'dest/sub/directory')
assert os.path.exists('dest/sub/directory/a/b/2')
def test_symlinks_true(self, stage):
"""Test installing with symlink preservation."""
with fs.working_dir(str(stage)):
fs.install_tree('source', 'dest', symlinks=True)
assert os.path.exists('dest/2')
assert os.path.islink('dest/2')
def test_symlinks_false(self, stage):
"""Test installing without symlink preservation."""
with fs.working_dir(str(stage)):
fs.install_tree('source', 'dest', symlinks=False)
assert os.path.exists('dest/2')
assert not os.path.islink('dest/2')
def test_move_transaction_commit(tmpdir):
fake_library = tmpdir.mkdir('lib').join('libfoo.so')
fake_library.write('Just some fake content.')
old_md5 = fs.hash_directory(str(tmpdir))
with fs.replace_directory_transaction(str(tmpdir.join('lib'))):
fake_library = tmpdir.mkdir('lib').join('libfoo.so')
fake_library.write('Other content.')
new_md5 = fs.hash_directory(str(tmpdir))
assert old_md5 != fs.hash_directory(str(tmpdir))
assert new_md5 == fs.hash_directory(str(tmpdir))
def test_move_transaction_rollback(tmpdir):
fake_library = tmpdir.mkdir('lib').join('libfoo.so')
fake_library.write('Just some fake content.')
h = fs.hash_directory(str(tmpdir))
try:
with fs.replace_directory_transaction(str(tmpdir.join('lib'))):
assert h != fs.hash_directory(str(tmpdir))
fake_library = tmpdir.mkdir('lib').join('libfoo.so')
fake_library.write('Other content.')
raise RuntimeError('')
except RuntimeError:
pass
assert h == fs.hash_directory(str(tmpdir))
|
