1 # GNU MediaGoblin -- federated, autonomous media hosting
2 # Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS.
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published by
6 # the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program. If not, see <http://www.gnu.org/licenses/>.
22 from werkzeug
.utils
import secure_filename
24 from mediagoblin
import storage
31 def test_clean_listy_filepath():
32 expected
= [u
'dir1', u
'dir2', u
'linooks.jpg']
33 assert storage
.clean_listy_filepath(
34 ['dir1', 'dir2', 'linooks.jpg']) == expected
36 expected
= [u
'dir1', u
'foo_.._nasty', u
'linooks.jpg']
37 assert storage
.clean_listy_filepath(
38 ['/dir1/', 'foo/../nasty', 'linooks.jpg']) == expected
40 expected
= [u
'etc', u
'passwd']
41 assert storage
.clean_listy_filepath(
42 ['../../../etc/', 'passwd']) == expected
44 with pytest
.raises(storage
.InvalidFilepath
):
45 storage
.clean_listy_filepath(['../../', 'linooks.jpg'])
48 class FakeStorageSystem():
49 def __init__(self
, foobie
, blech
, **kwargs
):
53 class FakeRemoteStorage(storage
.filestorage
.BasicFileStorage
):
54 # Theoretically despite this, all the methods should work but it
55 # should force copying to the workbench
58 def copy_local_to_storage(self
, *args
, **kwargs
):
59 return storage
.StorageInterface
.copy_local_to_storage(
60 self
, *args
, **kwargs
)
63 def test_storage_system_from_config():
64 this_storage
= storage
.storage_system_from_config(
65 {'base_url': 'http://example.org/moodia/',
67 'garbage_arg': 'garbage_arg',
68 'garbage_arg': 'trash'})
69 assert this_storage
.base_url
== 'http://example.org/moodia/'
70 assert this_storage
.base_dir
== '/tmp/'
71 assert this_storage
.__class
__ is storage
.filestorage
.BasicFileStorage
73 this_storage
= storage
.storage_system_from_config(
76 'garbage_arg': 'garbage_arg',
78 'mediagoblin.tests.test_storage:FakeStorageSystem'})
79 assert this_storage
.foobie
== 'eiboof'
80 assert this_storage
.blech
== 'hcelb'
81 assert unicode(this_storage
.__class
__) == \
82 u
'mediagoblin.tests.test_storage.FakeStorageSystem'
85 ##########################
86 # Basic file storage tests
87 ##########################
89 def get_tmp_filestorage(mount_url
=None, fake_remote
=False):
90 tmpdir
= tempfile
.mkdtemp(prefix
="test_gmg_storage")
92 this_storage
= FakeRemoteStorage(tmpdir
, mount_url
)
94 this_storage
= storage
.filestorage
.BasicFileStorage(tmpdir
, mount_url
)
95 return tmpdir
, this_storage
98 def cleanup_storage(this_storage
, tmpdir
, *paths
):
101 assert this_storage
.delete_dir(p
) == True
106 def test_basic_storage__resolve_filepath():
107 tmpdir
, this_storage
= get_tmp_filestorage()
109 result
= this_storage
._resolve
_filepath
(['dir1', 'dir2', 'filename.jpg'])
110 assert result
== os
.path
.join(
111 tmpdir
, 'dir1/dir2/filename.jpg')
113 result
= this_storage
._resolve
_filepath
(['../../etc/', 'passwd'])
114 assert result
== os
.path
.join(
115 tmpdir
, 'etc/passwd')
118 storage
.InvalidFilepath
,
119 this_storage
._resolve
_filepath
,
120 ['../../', 'etc', 'passwd'])
122 cleanup_storage(this_storage
, tmpdir
)
125 def test_basic_storage_file_exists():
126 tmpdir
, this_storage
= get_tmp_filestorage()
128 os
.makedirs(os
.path
.join(tmpdir
, 'dir1', 'dir2'))
129 filename
= os
.path
.join(tmpdir
, 'dir1', 'dir2', 'filename.txt')
130 with
open(filename
, 'w') as ourfile
:
131 ourfile
.write("I'm having a lovely day!")
133 assert this_storage
.file_exists(['dir1', 'dir2', 'filename.txt'])
134 assert not this_storage
.file_exists(['dir1', 'dir2', 'thisfile.lol'])
135 assert not this_storage
.file_exists(['dnedir1', 'dnedir2', 'somefile.lol'])
137 this_storage
.delete_file(['dir1', 'dir2', 'filename.txt'])
138 cleanup_storage(this_storage
, tmpdir
, ['dir1', 'dir2'])
141 def test_basic_storage_get_unique_filepath():
142 tmpdir
, this_storage
= get_tmp_filestorage()
144 # write something that exists
145 os
.makedirs(os
.path
.join(tmpdir
, 'dir1', 'dir2'))
146 filename
= os
.path
.join(tmpdir
, 'dir1', 'dir2', 'filename.txt')
147 with
open(filename
, 'w') as ourfile
:
148 ourfile
.write("I'm having a lovely day!")
150 # now we want something new, with the same name!
151 new_filepath
= this_storage
.get_unique_filepath(
152 ['dir1', 'dir2', 'filename.txt'])
153 assert new_filepath
[:-1] == [u
'dir1', u
'dir2']
155 new_filename
= new_filepath
[-1]
156 assert new_filename
.endswith('filename.txt')
157 assert len(new_filename
) > len('filename.txt')
158 assert new_filename
== secure_filename(new_filename
)
161 cleanup_storage(this_storage
, tmpdir
, ['dir1', 'dir2'])
164 def test_basic_storage_get_file():
165 tmpdir
, this_storage
= get_tmp_filestorage()
167 # Write a brand new file
168 filepath
= ['dir1', 'dir2', 'ourfile.txt']
170 with this_storage
.get_file(filepath
, 'w') as our_file
:
171 our_file
.write('First file')
172 with this_storage
.get_file(filepath
, 'r') as our_file
:
173 assert our_file
.read() == 'First file'
174 assert os
.path
.exists(os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
175 with
file(os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'), 'r') as our_file
:
176 assert our_file
.read() == 'First file'
178 # Write to the same path but try to get a unique file.
179 new_filepath
= this_storage
.get_unique_filepath(filepath
)
180 assert not os
.path
.exists(os
.path
.join(tmpdir
, *new_filepath
))
182 with this_storage
.get_file(new_filepath
, 'w') as our_file
:
183 our_file
.write('Second file')
184 with this_storage
.get_file(new_filepath
, 'r') as our_file
:
185 assert our_file
.read() == 'Second file'
186 assert os
.path
.exists(os
.path
.join(tmpdir
, *new_filepath
))
187 with
file(os
.path
.join(tmpdir
, *new_filepath
), 'r') as our_file
:
188 assert our_file
.read() == 'Second file'
190 # Read from an existing file
191 manually_written_file
= os
.makedirs(
192 os
.path
.join(tmpdir
, 'testydir'))
193 with
file(os
.path
.join(tmpdir
, 'testydir/testyfile.txt'), 'w') as testyfile
:
194 testyfile
.write('testy file! so testy.')
196 with this_storage
.get_file(['testydir', 'testyfile.txt']) as testyfile
:
197 assert testyfile
.read() == 'testy file! so testy.'
199 this_storage
.delete_file(filepath
)
200 this_storage
.delete_file(new_filepath
)
201 this_storage
.delete_file(['testydir', 'testyfile.txt'])
202 cleanup_storage(this_storage
, tmpdir
, ['dir1', 'dir2'], ['testydir'])
205 def test_basic_storage_delete_file():
206 tmpdir
, this_storage
= get_tmp_filestorage()
208 assert not os
.path
.exists(
209 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
211 filepath
= ['dir1', 'dir2', 'ourfile.txt']
212 with this_storage
.get_file(filepath
, 'w') as our_file
:
213 our_file
.write('Testing this file')
215 assert os
.path
.exists(
216 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
218 assert this_storage
.delete_dir(['dir1', 'dir2']) == False
219 this_storage
.delete_file(filepath
)
220 assert this_storage
.delete_dir(['dir1', 'dir2']) == True
222 assert not os
.path
.exists(
223 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
225 cleanup_storage(this_storage
, tmpdir
, ['dir1'])
228 def test_basic_storage_url_for_file():
229 # Not supplying a base_url should actually just bork.
230 tmpdir
, this_storage
= get_tmp_filestorage()
232 storage
.NoWebServing
,
233 this_storage
.file_url
,
234 ['dir1', 'dir2', 'filename.txt'])
235 cleanup_storage(this_storage
, tmpdir
)
237 # base_url without domain
238 tmpdir
, this_storage
= get_tmp_filestorage('/media/')
239 result
= this_storage
.file_url(
240 ['dir1', 'dir2', 'filename.txt'])
241 expected
= '/media/dir1/dir2/filename.txt'
242 assert result
== expected
243 cleanup_storage(this_storage
, tmpdir
)
245 # base_url with domain
246 tmpdir
, this_storage
= get_tmp_filestorage(
247 'http://media.example.org/ourmedia/')
248 result
= this_storage
.file_url(
249 ['dir1', 'dir2', 'filename.txt'])
250 expected
= 'http://media.example.org/ourmedia/dir1/dir2/filename.txt'
251 assert result
== expected
252 cleanup_storage(this_storage
, tmpdir
)
255 def test_basic_storage_get_local_path():
256 tmpdir
, this_storage
= get_tmp_filestorage()
258 result
= this_storage
.get_local_path(
259 ['dir1', 'dir2', 'filename.txt'])
261 expected
= os
.path
.join(
262 tmpdir
, 'dir1/dir2/filename.txt')
264 assert result
== expected
266 cleanup_storage(this_storage
, tmpdir
)
269 def test_basic_storage_is_local():
270 tmpdir
, this_storage
= get_tmp_filestorage()
271 assert this_storage
.local_storage
is True
272 cleanup_storage(this_storage
, tmpdir
)
275 def test_basic_storage_copy_locally():
276 tmpdir
, this_storage
= get_tmp_filestorage()
278 dest_tmpdir
= tempfile
.mkdtemp()
280 filepath
= ['dir1', 'dir2', 'ourfile.txt']
281 with this_storage
.get_file(filepath
, 'w') as our_file
:
282 our_file
.write('Testing this file')
284 new_file_dest
= os
.path
.join(dest_tmpdir
, 'file2.txt')
286 this_storage
.copy_locally(filepath
, new_file_dest
)
287 this_storage
.delete_file(filepath
)
289 assert file(new_file_dest
).read() == 'Testing this file'
291 os
.remove(new_file_dest
)
292 os
.rmdir(dest_tmpdir
)
293 cleanup_storage(this_storage
, tmpdir
, ['dir1', 'dir2'])
296 def _test_copy_local_to_storage_works(tmpdir
, this_storage
):
297 local_filename
= tempfile
.mktemp()
298 with
file(local_filename
, 'w') as tmpfile
:
299 tmpfile
.write('haha')
301 this_storage
.copy_local_to_storage(
302 local_filename
, ['dir1', 'dir2', 'copiedto.txt'])
304 os
.remove(local_filename
)
307 os
.path
.join(tmpdir
, 'dir1/dir2/copiedto.txt'),
308 'r').read() == 'haha'
310 this_storage
.delete_file(['dir1', 'dir2', 'copiedto.txt'])
311 cleanup_storage(this_storage
, tmpdir
, ['dir1', 'dir2'])
314 def test_basic_storage_copy_local_to_storage():
315 tmpdir
, this_storage
= get_tmp_filestorage()
316 _test_copy_local_to_storage_works(tmpdir
, this_storage
)
319 def test_general_storage_copy_local_to_storage():
320 tmpdir
, this_storage
= get_tmp_filestorage(fake_remote
=True)
321 _test_copy_local_to_storage_works(tmpdir
, this_storage
)