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/>.
17 ## Optional audio/video stuff
23 import gi
.repository
.Gst
24 # this gst initialization stuff is really required here
26 gi
.require_version('Gst', '1.0')
27 from gi
.repository
import Gst
29 from .media_tools
import create_av
35 import scikits
.audiolab
41 if six
.PY2
: # this hack only work in Python 2
44 sys
.setdefaultencoding('utf-8')
51 import six
.moves
.urllib
.parse
as urlparse
53 from mediagoblin
.tests
.tools
import (
54 fixture_add_user
, fixture_add_collection
, get_app
)
55 from mediagoblin
import mg_globals
56 from mediagoblin
.db
.models
import MediaEntry
, User
, LocalUser
, Activity
57 from mediagoblin
.db
.base
import Session
58 from mediagoblin
.tools
import template
59 from mediagoblin
.media_types
.image
import ImageMediaManager
60 from mediagoblin
.media_types
.pdf
.processing
import check_prerequisites
as pdf_check_prerequisites
62 from .resources
import GOOD_JPG
, GOOD_PNG
, EVIL_FILE
, EVIL_JPG
, EVIL_PNG
, \
63 BIG_BLUE
, GOOD_PDF
, GPS_JPG
, MED_PNG
, BIG_PNG
65 GOOD_TAG_STRING
= u
'yin,yang'
66 BAD_TAG_STRING
= six
.text_type('rage,' + 'f' * 26 + 'u' * 26)
68 FORM_CONTEXT
= ['mediagoblin/submit/start.html', 'submit_form']
69 REQUEST_CONTEXT
= ['mediagoblin/user_pages/user.html', 'request']
73 def audio_plugin_app(request
):
76 mgoblin_config
=pkg_resources
.resource_filename(
78 'test_mgoblin_app_audio.ini'))
81 def video_plugin_app(request
):
84 mgoblin_config
=pkg_resources
.resource_filename(
86 'test_mgoblin_app_video.ini'))
89 def audio_video_plugin_app(request
):
92 mgoblin_config
=pkg_resources
.resource_filename(
94 'test_mgoblin_app_audio_video.ini'))
97 def pdf_plugin_app(request
):
100 mgoblin_config
=pkg_resources
.resource_filename(
102 'test_mgoblin_app_pdf.ini'))
105 class BaseTestSubmission
:
106 @pytest.fixture(autouse
=True)
107 def setup(self
, test_app
):
108 self
.test_app
= test_app
110 # TODO: Possibly abstract into a decorator like:
111 # @as_authenticated_user('chris')
112 fixture_add_user(privileges
=[u
'active',u
'uploader', u
'commenter'])
118 Fetch the user we're submitting with. Every .get() or .post()
119 invalidates the session; this is a hacky workaround.
121 #### FIXME: Pytest collects this as a test and runs this.
122 #### ... it shouldn't. At least it passes, but that's
124 #### Also if we found a way to make this run it should be a
126 return LocalUser
.query
.filter(LocalUser
.username
==u
'chris').first()
131 'username': u
'chris',
132 'password': 'toast'})
135 self
.test_app
.get('/auth/logout/')
137 def do_post(self
, data
, *context_keys
, **kwargs
):
138 url
= kwargs
.pop('url', '/submit/')
139 do_follow
= kwargs
.pop('do_follow', False)
140 template
.clear_test_template_context()
141 response
= self
.test_app
.post(url
, data
, **kwargs
)
144 context_data
= template
.TEMPLATE_TEST_CONTEXT
145 for key
in context_keys
:
146 context_data
= context_data
[key
]
147 return response
, context_data
149 def upload_data(self
, filename
):
150 return {'upload_files': [('file', filename
)]}
152 def check_comments(self
, request
, media_id
, count
):
153 gmr
= request
.db
.GenericModelReference
.query
.filter_by(
155 model_type
=request
.db
.MediaEntry
.__tablename
__
157 if gmr
is None and count
<= 0:
158 return # Yerp it's fine.
159 comments
= request
.db
.Comment
.query
.filter_by(target_id
=gmr
.id)
160 assert count
== comments
.count()
162 def check_url(self
, response
, path
):
163 assert urlparse
.urlsplit(response
.location
)[2] == path
165 def check_normal_upload(self
, title
, filename
):
166 response
, context
= self
.do_post({'title': title
}, do_follow
=True,
167 **self
.upload_data(filename
))
168 self
.check_url(response
, '/u/{0}/'.format(self
.our_user().username
))
169 assert 'mediagoblin/user_pages/user.html' in context
170 # Make sure the media view is at least reachable, logged in...
171 url
= '/u/{0}/m/{1}/'.format(self
.our_user().username
,
172 title
.lower().replace(' ', '-'))
173 self
.test_app
.get(url
)
174 # ... and logged out too.
176 self
.test_app
.get(url
)
178 def user_upload_limits(self
, uploaded
=None, upload_limit
=None):
179 our_user
= self
.our_user()
182 our_user
.uploaded
= uploaded
184 our_user
.upload_limit
= upload_limit
187 Session
.expunge(our_user
)
190 class TestSubmissionBasics(BaseTestSubmission
):
191 def test_missing_fields(self
):
194 response
, form
= self
.do_post({}, *FORM_CONTEXT
)
195 assert form
.file.errors
== [u
'You must provide a file.']
199 response
, form
= self
.do_post({'title': u
'test title'}, *FORM_CONTEXT
)
200 assert form
.file.errors
== [u
'You must provide a file.']
202 def test_normal_jpg(self
):
203 # User uploaded should be 0
204 assert self
.our_user().uploaded
== 0
206 self
.check_normal_upload(u
'Normal upload 1', GOOD_JPG
)
208 # User uploaded should be the same as GOOD_JPG size in Mb
209 file_size
= os
.stat(GOOD_JPG
).st_size
/ (1024.0 * 1024)
210 file_size
= float('{0:.2f}'.format(file_size
))
213 assert self
.our_user().uploaded
== file_size
215 def test_public_id_populated(self
):
216 # Upload the image first.
217 response
, request
= self
.do_post({'title': u
'Balanced Goblin'},
218 *REQUEST_CONTEXT
, do_follow
=True,
219 **self
.upload_data(GOOD_JPG
))
220 media
= self
.check_media(request
, {'title': u
'Balanced Goblin'}, 1)
222 # Now check that the public_id attribute is set.
223 assert media
.public_id
!= None
225 def test_normal_png(self
):
226 self
.check_normal_upload(u
'Normal upload 2', GOOD_PNG
)
228 def test_default_upload_limits(self
):
229 self
.user_upload_limits(uploaded
=500)
231 # User uploaded should be 500
232 assert self
.our_user().uploaded
== 500
234 response
, context
= self
.do_post({'title': u
'Normal upload 4'},
236 **self
.upload_data(GOOD_JPG
))
237 self
.check_url(response
, '/u/{0}/'.format(self
.our_user().username
))
238 assert 'mediagoblin/user_pages/user.html' in context
240 # Shouldn't have uploaded
241 assert self
.our_user().uploaded
== 500
243 def test_user_upload_limit(self
):
244 self
.user_upload_limits(uploaded
=25, upload_limit
=25)
246 # User uploaded should be 25
247 assert self
.our_user().uploaded
== 25
249 response
, context
= self
.do_post({'title': u
'Normal upload 5'},
251 **self
.upload_data(GOOD_JPG
))
252 self
.check_url(response
, '/u/{0}/'.format(self
.our_user().username
))
253 assert 'mediagoblin/user_pages/user.html' in context
255 # Shouldn't have uploaded
256 assert self
.our_user().uploaded
== 25
258 def test_user_under_limit(self
):
259 self
.user_upload_limits(uploaded
=499)
261 # User uploaded should be 499
262 assert self
.our_user().uploaded
== 499
264 response
, context
= self
.do_post({'title': u
'Normal upload 6'},
266 **self
.upload_data(MED_PNG
))
267 form
= context
['mediagoblin/submit/start.html']['submit_form']
268 assert form
.file.errors
== [u
'Sorry, uploading this file will put you'
269 ' over your upload limit.']
271 # Shouldn't have uploaded
272 assert self
.our_user().uploaded
== 499
274 def test_big_file(self
):
275 response
, context
= self
.do_post({'title': u
'Normal upload 7'},
277 **self
.upload_data(BIG_PNG
))
279 form
= context
['mediagoblin/submit/start.html']['submit_form']
280 assert form
.file.errors
== [u
'Sorry, the file size is too big.']
282 def check_media(self
, request
, find_data
, count
=None):
283 media
= MediaEntry
.query
.filter_by(**find_data
)
284 if count
is not None:
285 assert media
.count() == count
293 response
, request
= self
.do_post({'title': u
'Balanced Goblin 2',
294 'tags': GOOD_TAG_STRING
},
295 *REQUEST_CONTEXT
, do_follow
=True,
296 **self
.upload_data(GOOD_JPG
))
297 media
= self
.check_media(request
, {'title': u
'Balanced Goblin 2'}, 1)
298 assert media
.tags
[0]['name'] == u
'yin'
299 assert media
.tags
[0]['slug'] == u
'yin'
301 assert media
.tags
[1]['name'] == u
'yang'
302 assert media
.tags
[1]['slug'] == u
'yang'
304 # Test tags that are too long
306 response
, form
= self
.do_post({'title': u
'Balanced Goblin 2',
307 'tags': BAD_TAG_STRING
},
309 **self
.upload_data(GOOD_JPG
))
310 assert form
.tags
.errors
== [
311 u
'Tags must be shorter than 50 characters. ' \
312 'Tags that are too long: ' \
313 'ffffffffffffffffffffffffffuuuuuuuuuuuuuuuuuuuuuuuuuu']
315 def test_delete(self
):
316 self
.user_upload_limits(uploaded
=50)
317 response
, request
= self
.do_post({'title': u
'Balanced Goblin'},
318 *REQUEST_CONTEXT
, do_follow
=True,
319 **self
.upload_data(GOOD_JPG
))
320 media
= self
.check_media(request
, {'title': u
'Balanced Goblin'}, 1)
323 # render and post to the edit page.
324 edit_url
= request
.urlgen(
325 'mediagoblin.edit.edit_media',
326 user
=self
.our_user().username
, media_id
=media_id
)
327 self
.test_app
.get(edit_url
)
328 self
.test_app
.post(edit_url
,
329 {'title': u
'Balanced Goblin',
330 'slug': u
"Balanced=Goblin",
332 media
= self
.check_media(request
, {'title': u
'Balanced Goblin'}, 1)
333 assert media
.slug
== u
"balanced-goblin"
335 # Add a comment, so we can test for its deletion later.
336 self
.check_comments(request
, media_id
, 0)
337 comment_url
= request
.urlgen(
338 'mediagoblin.user_pages.media_post_comment',
339 user
=self
.our_user().username
, media_id
=media_id
)
340 response
= self
.do_post({'comment_content': 'i love this test'},
341 url
=comment_url
, do_follow
=True)[0]
342 self
.check_comments(request
, media_id
, 1)
344 # Do not confirm deletion
345 # ---------------------------------------------------
346 delete_url
= request
.urlgen(
347 'mediagoblin.user_pages.media_confirm_delete',
348 user
=self
.our_user().username
, media_id
=media_id
)
349 # Empty data means don't confirm
350 response
= self
.do_post({}, do_follow
=True, url
=delete_url
)[0]
351 media
= self
.check_media(request
, {'title': u
'Balanced Goblin'}, 1)
355 # ---------------------------------------------------
356 response
, request
= self
.do_post({'confirm': 'y'}, *REQUEST_CONTEXT
,
357 do_follow
=True, url
=delete_url
)
358 self
.check_media(request
, {'id': media_id
}, 0)
359 self
.check_comments(request
, media_id
, 0)
361 # Check that user.uploaded is the same as before the upload
362 assert self
.our_user().uploaded
== 50
364 def test_evil_file(self
):
365 # Test non-suppoerted file with non-supported extension
366 # -----------------------------------------------------
367 response
, form
= self
.do_post({'title': u
'Malicious Upload 1'},
369 **self
.upload_data(EVIL_FILE
))
370 assert len(form
.file.errors
) == 1
371 assert 'Sorry, I don\'t support that file type :(' == \
372 str(form
.file.errors
[0])
375 def test_get_media_manager(self
):
376 """Test if the get_media_manger function returns sensible things
378 response
, request
= self
.do_post({'title': u
'Balanced Goblin'},
379 *REQUEST_CONTEXT
, do_follow
=True,
380 **self
.upload_data(GOOD_JPG
))
381 media
= self
.check_media(request
, {'title': u
'Balanced Goblin'}, 1)
383 assert media
.media_type
== u
'mediagoblin.media_types.image'
384 assert isinstance(media
.media_manager
, ImageMediaManager
)
385 assert media
.media_manager
.entry
== media
388 def test_sniffing(self
):
390 Test sniffing mechanism to assert that regular uploads work as intended
392 template
.clear_test_template_context()
393 response
= self
.test_app
.post(
395 'title': u
'UNIQUE_TITLE_PLS_DONT_CREATE_OTHER_MEDIA_WITH_THIS_TITLE'
401 context
= template
.TEMPLATE_TEST_CONTEXT
['mediagoblin/user_pages/user.html']
403 request
= context
['request']
405 media
= request
.db
.MediaEntry
.query
.filter_by(
406 title
=u
'UNIQUE_TITLE_PLS_DONT_CREATE_OTHER_MEDIA_WITH_THIS_TITLE').first()
408 assert media
.media_type
== 'mediagoblin.media_types.image'
410 def check_false_image(self
, title
, filename
):
411 # NOTE: The following 2 tests will ultimately fail, but they
412 # *will* pass the initial form submission step. Instead,
413 # they'll be caught as failures during the processing step.
414 response
, context
= self
.do_post({'title': title
}, do_follow
=True,
415 **self
.upload_data(filename
))
416 self
.check_url(response
, '/u/{0}/'.format(self
.our_user().username
))
417 entry
= mg_globals
.database
.MediaEntry
.query
.filter_by(title
=title
).first()
418 assert entry
.state
== 'failed'
419 assert entry
.fail_error
== u
'mediagoblin.processing:BadMediaFail'
421 def test_evil_jpg(self
):
422 # Test non-supported file with .jpg extension
423 # -------------------------------------------
424 self
.check_false_image(u
'Malicious Upload 2', EVIL_JPG
)
426 def test_evil_png(self
):
427 # Test non-supported file with .png extension
428 # -------------------------------------------
429 self
.check_false_image(u
'Malicious Upload 3', EVIL_PNG
)
431 def test_media_data(self
):
432 self
.check_normal_upload(u
"With GPS data", GPS_JPG
)
433 media
= self
.check_media(None, {"title": u
"With GPS data"}, 1)
434 assert media
.get_location
.position
["latitude"] == 59.336666666666666
436 def test_processing(self
):
437 public_store_dir
= mg_globals
.global_config
[
438 'storage:publicstore']['base_dir']
440 data
= {'title': u
'Big Blue'}
441 response
, request
= self
.do_post(data
, *REQUEST_CONTEXT
, do_follow
=True,
442 **self
.upload_data(BIG_BLUE
))
443 media
= self
.check_media(request
, data
, 1)
444 last_size
= 1024 ** 3 # Needs to be larger than bigblue.png
445 for key
, basename
in (('original', 'bigblue.png'),
446 ('medium', 'bigblue.medium.png'),
447 ('thumb', 'bigblue.thumbnail.png')):
448 # Does the processed image have a good filename?
449 filename
= os
.path
.join(
451 *media
.media_files
[key
])
452 assert filename
.endswith('_' + basename
)
453 # Is it smaller than the last processed image we looked at?
454 size
= os
.stat(filename
).st_size
455 assert last_size
> size
458 def test_collection_selection(self
):
459 """Test the ability to choose a collection when submitting media
461 # Collection option should have been removed if the user has no
463 response
= self
.test_app
.get('/submit/')
464 assert 'collection' not in response
.form
.fields
466 # Test upload of an image when a user has no collections.
467 upload
= webtest
.forms
.Upload(os
.path
.join(
468 'mediagoblin', 'static', 'images', 'media_thumbs', 'image.png'))
469 response
.form
['file'] = upload
470 no_collection_title
= 'no collection'
471 response
.form
['title'] = no_collection_title
472 response
.form
.submit()
473 assert MediaEntry
.query
.filter_by(
474 actor
=self
.our_user().id
475 ).first().title
== no_collection_title
477 # Collection option should be present if the user has collections. It
478 # shouldn't allow other users' collections to be selected.
479 col
= fixture_add_collection(user
=self
.our_user())
480 user
= fixture_add_user(username
=u
'different')
481 fixture_add_collection(user
=user
, name
=u
'different')
482 response
= self
.test_app
.get('/submit/')
484 assert 'collection' in form
.fields
485 # Option length is 2, because of the default "--Select--" option
486 assert len(form
['collection'].options
) == 2
487 assert form
['collection'].options
[1][2] == col
.title
489 # Test that if we specify a collection then the media entry is added to
490 # the specified collection.
491 form
['file'] = upload
492 title
= 'new picture'
493 form
['title'] = title
494 form
['collection'] = form
['collection'].options
[1][0]
496 # The title of the first item in our user's first collection should
497 # match the title of the picture that was just added.
498 col
= self
.our_user().collections
[0]
499 assert col
.collection_items
[0].get_object().title
== title
501 # Test that an activity was created when the item was added to the
502 # collection. That should be the last activity.
503 assert Activity
.query
.order_by(
505 ).first().content
== '{0} added new picture to {1}'.format(
506 self
.our_user().username
, col
.title
)
508 # Test upload succeeds if the user has collection and no collection is
510 form
['file'] = webtest
.forms
.Upload(os
.path
.join(
511 'mediagoblin', 'static', 'images', 'media_thumbs', 'image.png'))
512 title
= 'no collection 2'
513 form
['title'] = title
514 form
['collection'] = form
['collection'].options
[0][0]
516 # The title of the first item in our user's first collection should
517 # match the title of the picture that was just added.
518 assert MediaEntry
.query
.filter_by(
519 actor
=self
.our_user().id
522 class TestSubmissionVideo(BaseTestSubmission
):
523 @pytest.fixture(autouse
=True)
524 def setup(self
, video_plugin_app
):
525 self
.test_app
= video_plugin_app
527 # TODO: Possibly abstract into a decorator like:
528 # @as_authenticated_user('chris')
529 fixture_add_user(privileges
=[u
'active',u
'uploader', u
'commenter'])
533 @pytest.mark
.skipif(SKIP_VIDEO
,
534 reason
="Dependencies for video not met")
535 def test_video(self
, video_plugin_app
):
536 with
create_av(make_video
=True) as path
:
537 self
.check_normal_upload('Video', path
)
540 class TestSubmissionAudio(BaseTestSubmission
):
541 @pytest.fixture(autouse
=True)
542 def setup(self
, audio_plugin_app
):
543 self
.test_app
= audio_plugin_app
545 # TODO: Possibly abstract into a decorator like:
546 # @as_authenticated_user('chris')
547 fixture_add_user(privileges
=[u
'active',u
'uploader', u
'commenter'])
551 @pytest.mark
.skipif(SKIP_AUDIO
,
552 reason
="Dependencies for audio not met")
553 def test_audio(self
, audio_plugin_app
):
554 with
create_av(make_audio
=True) as path
:
555 self
.check_normal_upload('Audio', path
)
558 class TestSubmissionAudioVideo(BaseTestSubmission
):
559 @pytest.fixture(autouse
=True)
560 def setup(self
, audio_video_plugin_app
):
561 self
.test_app
= audio_video_plugin_app
563 # TODO: Possibly abstract into a decorator like:
564 # @as_authenticated_user('chris')
565 fixture_add_user(privileges
=[u
'active',u
'uploader', u
'commenter'])
569 @pytest.mark
.skipif(SKIP_AUDIO
or SKIP_VIDEO
,
570 reason
="Dependencies for audio or video not met")
571 def test_audio_and_video(self
):
572 with
create_av(make_audio
=True, make_video
=True) as path
:
573 self
.check_normal_upload('Audio and Video', path
)
576 class TestSubmissionPDF(BaseTestSubmission
):
577 @pytest.fixture(autouse
=True)
578 def setup(self
, pdf_plugin_app
):
579 self
.test_app
= pdf_plugin_app
581 # TODO: Possibly abstract into a decorator like:
582 # @as_authenticated_user('chris')
583 fixture_add_user(privileges
=[u
'active',u
'uploader', u
'commenter'])
587 @pytest.mark
.skipif("not os.path.exists(GOOD_PDF) or not pdf_check_prerequisites()")
588 def test_normal_pdf(self
):
589 response
, context
= self
.do_post({'title': u
'Normal upload 3 (pdf)'},
591 **self
.upload_data(GOOD_PDF
))
592 self
.check_url(response
, '/u/{0}/'.format(self
.our_user().username
))
593 assert 'mediagoblin/user_pages/user.html' in context