added a/v submission testing

mediagoblin/tests/media_tools.py

@@ -0,0 +1,61 @@
+# GNU MediaGoblin -- federated, autonomous media hosting
+# Copyright (C) 2013 MediaGoblin contributors.  See AUTHORS.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from contextlib import contextmanager
+import tempfile
+
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import Gst
+Gst.init(None)
+
+@contextmanager
+def create_av(make_video=False, make_audio=False):
+    'creates audio/video in `path`, throws AssertionError on any error'
+    media = tempfile.NamedTemporaryFile(suffix='.ogg')
+    pipeline = Gst.Pipeline()
+    mux = Gst.ElementFactory.make('oggmux', 'mux')
+    pipeline.add(mux)
+    if make_video:
+        video_src = Gst.ElementFactory.make('videotestsrc', 'video_src')
+        video_src.set_property('num-buffers', 20)
+        video_enc = Gst.ElementFactory.make('theoraenc', 'video_enc')
+        pipeline.add(video_src)
+        pipeline.add(video_enc)
+        assert video_src.link(video_enc)
+        assert video_enc.link(mux)
+    if make_audio:
+        audio_src = Gst.ElementFactory.make('audiotestsrc', 'audio_src')
+        audio_src.set_property('num-buffers', 20)
+        audio_enc = Gst.ElementFactory.make('vorbisenc', 'audio_enc')
+        pipeline.add(audio_src)
+        pipeline.add(audio_enc)
+        assert audio_src.link(audio_enc)
+        assert audio_enc.link(mux)
+    sink = Gst.ElementFactory.make('filesink', 'sink')
+    sink.set_property('location', media.name)
+    pipeline.add(sink)
+    mux.link(sink)
+    pipeline.set_state(Gst.State.PLAYING)
+    state = pipeline.get_state(Gst.SECOND)
+    assert state[0] == Gst.StateChangeReturn.SUCCESS
+    bus = pipeline.get_bus()
+    message = bus.timed_pop_filtered(
+            Gst.SECOND,  # one second should be more than enough for 50-buf vid
+            Gst.MessageType.ERROR | Gst.MessageType.EOS)
+    assert message.type == Gst.MessageType.EOS
+    pipeline.set_state(Gst.State.NULL)
+    yield media.name

mediagoblin/tests/test_mgoblin_app.ini

@@ -37,4 +37,6 @@ BROKER_URL = "sqlite:///%(here)s/test_user_dev/kombu.db"
 [[mediagoblin.plugins.basic_auth]]
 [[mediagoblin.plugins.openid]]
 [[mediagoblin.media_types.image]]
+[[mediagoblin.media_types.video]]
+[[mediagoblin.media_types.audio]]
 [[mediagoblin.media_types.pdf]]

mediagoblin/tests/test_submission.py

@@ -26,7 +26,14 @@ import pytest
 
 import six.moves.urllib.parse as urlparse
 
+# this gst initialization stuff is really required here
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import Gst
+Gst.init(None)
+
 from mediagoblin.tests.tools import fixture_add_user
+from .media_tools import create_av
 from mediagoblin import mg_globals
 from mediagoblin.db.models import MediaEntry, User
 from mediagoblin.db.base import Session
@@ -365,6 +372,18 @@ class TestSubmission:
         media = self.check_media(None, {"title": u"With GPS data"}, 1)
         assert media.get_location.position["latitude"] == 59.336666666666666
 
+    def test_audio(self):
+        with create_av(make_audio=True) as path:
+            self.check_normal_upload('Audio', path)
+
+    def test_video(self):
+        with create_av(make_video=True) as path:
+            self.check_normal_upload('Video', path)
+
+    def test_audio_and_video(self):
+        with create_av(make_audio=True, make_video=True) as path:
+            self.check_normal_upload('Audio and Video', path)
+
     def test_processing(self):
         public_store_dir = mg_globals.global_config[
             'storage:publicstore']['base_dir']