55b66e84ea990e758025cb5590d5da3be63ed9d0
1 # GNU MediaGoblin -- federated, autonomous media hosting
2 # Copyright (C) 2011 Free Software Foundation, Inc
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/>.
21 from nose
.tools
import assert_raises
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
45 storage
.InvalidFilepath
,
46 storage
.clean_listy_filepath
,
47 ['../../', 'linooks.jpg'])
50 class FakeStorageSystem():
51 def __init__(self
, foobie
, blech
, **kwargs
):
55 class FakeRemoteStorage(storage
.BasicFileStorage
):
56 # Theoretically despite this, all the methods should work but it
57 # should force copying to the workbench
61 def test_storage_system_from_paste_config():
62 this_storage
= storage
.storage_system_from_paste_config(
63 {'somestorage_base_url': 'http://example.org/moodia/',
64 'somestorage_base_dir': '/tmp/',
65 'somestorage_garbage_arg': 'garbage_arg',
66 'garbage_arg': 'trash'},
68 assert this_storage
.base_url
== 'http://example.org/moodia/'
69 assert this_storage
.base_dir
== '/tmp/'
70 assert this_storage
.__class
__ is storage
.BasicFileStorage
72 this_storage
= storage
.storage_system_from_paste_config(
73 {'somestorage_foobie': 'eiboof',
74 'somestorage_blech': 'hcelb',
75 'somestorage_garbage_arg': 'garbage_arg',
76 'garbage_arg': 'trash',
77 'somestorage_storage_class':
78 'mediagoblin.tests.test_storage:FakeStorageSystem'},
80 assert this_storage
.foobie
== 'eiboof'
81 assert this_storage
.blech
== 'hcelb'
82 assert this_storage
.__class
__ is FakeStorageSystem
85 ##########################
86 # Basic file storage tests
87 ##########################
89 def get_tmp_filestorage(mount_url
=None, fake_remote
=False):
90 tmpdir
= tempfile
.mkdtemp()
92 this_storage
= FakeRemoteStorage(tmpdir
, mount_url
)
94 this_storage
= storage
.BasicFileStorage(tmpdir
, mount_url
)
95 return tmpdir
, this_storage
98 def test_basic_storage__resolve_filepath():
99 tmpdir
, this_storage
= get_tmp_filestorage()
101 result
= this_storage
._resolve
_filepath
(['dir1', 'dir2', 'filename.jpg'])
102 assert result
== os
.path
.join(
103 tmpdir
, 'dir1/dir2/filename.jpg')
105 result
= this_storage
._resolve
_filepath
(['../../etc/', 'passwd'])
106 assert result
== os
.path
.join(
107 tmpdir
, 'etc/passwd')
110 storage
.InvalidFilepath
,
111 this_storage
._resolve
_filepath
,
112 ['../../', 'etc', 'passwd'])
115 def test_basic_storage_file_exists():
116 tmpdir
, this_storage
= get_tmp_filestorage()
118 os
.makedirs(os
.path
.join(tmpdir
, 'dir1', 'dir2'))
119 filename
= os
.path
.join(tmpdir
, 'dir1', 'dir2', 'filename.txt')
120 with
open(filename
, 'w') as ourfile
:
121 ourfile
.write("I'm having a lovely day!")
123 assert this_storage
.file_exists(['dir1', 'dir2', 'filename.txt'])
124 assert not this_storage
.file_exists(['dir1', 'dir2', 'thisfile.lol'])
125 assert not this_storage
.file_exists(['dnedir1', 'dnedir2', 'somefile.lol'])
128 def test_basic_storage_get_unique_filepath():
129 tmpdir
, this_storage
= get_tmp_filestorage()
131 # write something that exists
132 os
.makedirs(os
.path
.join(tmpdir
, 'dir1', 'dir2'))
133 filename
= os
.path
.join(tmpdir
, 'dir1', 'dir2', 'filename.txt')
134 with
open(filename
, 'w') as ourfile
:
135 ourfile
.write("I'm having a lovely day!")
137 # now we want something new, with the same name!
138 new_filepath
= this_storage
.get_unique_filepath(
139 ['dir1', 'dir2', 'filename.txt'])
140 assert new_filepath
[:-1] == [u
'dir1', u
'dir2']
142 new_filename
= new_filepath
[-1]
143 assert new_filename
.endswith('filename.txt')
144 assert len(new_filename
) > len('filename.txt')
145 assert new_filename
== secure_filename(new_filename
)
148 def test_basic_storage_get_file():
149 tmpdir
, this_storage
= get_tmp_filestorage()
151 # Write a brand new file
152 filepath
= ['dir1', 'dir2', 'ourfile.txt']
154 with this_storage
.get_file(filepath
, 'w') as our_file
:
155 our_file
.write('First file')
156 with this_storage
.get_file(filepath
, 'r') as our_file
:
157 assert our_file
.read() == 'First file'
158 assert os
.path
.exists(os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
159 with
file(os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'), 'r') as our_file
:
160 assert our_file
.read() == 'First file'
162 # Write to the same path but try to get a unique file.
163 new_filepath
= this_storage
.get_unique_filepath(filepath
)
164 assert not os
.path
.exists(os
.path
.join(tmpdir
, *new_filepath
))
166 with this_storage
.get_file(new_filepath
, 'w') as our_file
:
167 our_file
.write('Second file')
168 with this_storage
.get_file(new_filepath
, 'r') as our_file
:
169 assert our_file
.read() == 'Second file'
170 assert os
.path
.exists(os
.path
.join(tmpdir
, *new_filepath
))
171 with
file(os
.path
.join(tmpdir
, *new_filepath
), 'r') as our_file
:
172 assert our_file
.read() == 'Second file'
174 # Read from an existing file
175 manually_written_file
= os
.makedirs(
176 os
.path
.join(tmpdir
, 'testydir'))
177 with
file(os
.path
.join(tmpdir
, 'testydir/testyfile.txt'), 'w') as testyfile
:
178 testyfile
.write('testy file! so testy.')
180 with this_storage
.get_file(['testydir', 'testyfile.txt']) as testyfile
:
181 assert testyfile
.read() == 'testy file! so testy.'
184 def test_basic_storage_delete_file():
185 tmpdir
, this_storage
= get_tmp_filestorage()
187 assert not os
.path
.exists(
188 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
190 filepath
= ['dir1', 'dir2', 'ourfile.txt']
191 with this_storage
.get_file(filepath
, 'w') as our_file
:
192 our_file
.write('Testing this file')
194 assert os
.path
.exists(
195 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
197 this_storage
.delete_file(filepath
)
199 assert not os
.path
.exists(
200 os
.path
.join(tmpdir
, 'dir1/dir2/ourfile.txt'))
203 def test_basic_storage_url_for_file():
204 # Not supplying a base_url should actually just bork.
205 tmpdir
, this_storage
= get_tmp_filestorage()
207 storage
.NoWebServing
,
208 this_storage
.file_url
,
209 ['dir1', 'dir2', 'filename.txt'])
211 # base_url without domain
212 tmpdir
, this_storage
= get_tmp_filestorage('/media/')
213 result
= this_storage
.file_url(
214 ['dir1', 'dir2', 'filename.txt'])
215 expected
= '/media/dir1/dir2/filename.txt'
216 assert result
== expected
218 # base_url with domain
219 tmpdir
, this_storage
= get_tmp_filestorage(
220 'http://media.example.org/ourmedia/')
221 result
= this_storage
.file_url(
222 ['dir1', 'dir2', 'filename.txt'])
223 expected
= 'http://media.example.org/ourmedia/dir1/dir2/filename.txt'
224 assert result
== expected
227 def test_basic_storage_get_local_path():
228 tmpdir
, this_storage
= get_tmp_filestorage()
230 result
= this_storage
.get_local_path(
231 ['dir1', 'dir2', 'filename.txt'])
233 expected
= os
.path
.join(
234 tmpdir
, 'dir1/dir2/filename.txt')
236 assert result
== expected
239 def test_basic_storage_is_local():
240 tmpdir
, this_storage
= get_tmp_filestorage()
241 assert this_storage
.local_storage
is True
244 def test_basic_storage_copy_locally():
245 tmpdir
, this_storage
= get_tmp_filestorage()
247 dest_tmpdir
= tempfile
.mkdtemp()
249 filepath
= ['dir1', 'dir2', 'ourfile.txt']
250 with this_storage
.get_file(filepath
, 'w') as our_file
:
251 our_file
.write('Testing this file')
253 new_file_dest
= os
.path
.join(dest_tmpdir
, 'file2.txt')
255 this_storage
.copy_locally(filepath
, new_file_dest
)
257 assert file(new_file_dest
).read() == 'Testing this file'