* morituri/common/common.py:
add quoteParse function to quote properly for gst.parse_launch() * morituri/common/checksum.py: Fix single quote problem. use self.setException() * morituri/common/encode.py: Fix single quote problem. * morituri/image/image.py: Fix single quote problem. Reraise gst.QueryError. * morituri/test/test_common_encode.py: * morituri/test/test_common_encode.py (added): Add tests for unicode/single/double quote. * morituri/test/test_image_image.py: * morituri/test/test_common_checksum.py: Add tests for single/double quote. * morituri/test/common.py: add twisted-like failUnlessRaises to TestCase that returns the exception, so we can check wrapped exceptions in TaskException.
This commit is contained in:
22
ChangeLog
22
ChangeLog
@@ -1,3 +1,25 @@
|
||||
2010-04-05 Thomas Vander Stichele <thomas at apestaart dot org>
|
||||
|
||||
* morituri/common/common.py:
|
||||
add quoteParse function to quote properly for gst.parse_launch()
|
||||
* morituri/common/checksum.py:
|
||||
Fix single quote problem.
|
||||
use self.setException()
|
||||
* morituri/common/encode.py:
|
||||
Fix single quote problem.
|
||||
* morituri/image/image.py:
|
||||
Fix single quote problem.
|
||||
Reraise gst.QueryError.
|
||||
* morituri/test/test_common_encode.py:
|
||||
* morituri/test/test_common_encode.py (added):
|
||||
Add tests for unicode/single/double quote.
|
||||
* morituri/test/test_image_image.py:
|
||||
* morituri/test/test_common_checksum.py:
|
||||
Add tests for single/double quote.
|
||||
* morituri/test/common.py:
|
||||
add twisted-like failUnlessRaises to TestCase that returns
|
||||
the exception, so we can check wrapped exceptions in TaskException.
|
||||
|
||||
2010-04-05 Thomas Vander Stichele <thomas at apestaart dot org>
|
||||
|
||||
* morituri/common/task.py:
|
||||
|
||||
@@ -76,7 +76,7 @@ class ChecksumTask(task.Task):
|
||||
filesrc location="%s" !
|
||||
decodebin ! audio/x-raw-int !
|
||||
appsink name=sink sync=False emit-signals=True''' %
|
||||
self._path.encode('utf-8'))
|
||||
common.quoteParse(self._path).encode('utf-8'))
|
||||
|
||||
self.debug('pausing pipeline')
|
||||
self._pipeline.set_state(gst.STATE_PAUSED)
|
||||
@@ -90,7 +90,7 @@ class ChecksumTask(task.Task):
|
||||
try:
|
||||
length, qformat = sink.query_duration(gst.FORMAT_DEFAULT)
|
||||
except gst.QueryError, e:
|
||||
self.exception = e
|
||||
self.setException(e)
|
||||
self.stop()
|
||||
return
|
||||
|
||||
@@ -187,7 +187,7 @@ class ChecksumTask(task.Task):
|
||||
|
||||
if not self._last:
|
||||
# see http://bugzilla.gnome.org/show_bug.cgi?id=578612
|
||||
print 'ERROR: not a single buffer gotten'
|
||||
print 'ERROR: checksum: not a single buffer gotten'
|
||||
# FIXME: instead of print, do something useful
|
||||
else:
|
||||
self._checksum = self._checksum % 2 ** 32
|
||||
|
||||
@@ -67,6 +67,15 @@ def framesToHMSF(frames):
|
||||
|
||||
return "%02d:%02d:%02d.%02d" % (h, m, s, f)
|
||||
|
||||
def quoteParse(path):
|
||||
"""
|
||||
Quote a path for use in gst.parse_launch.
|
||||
"""
|
||||
# Make sure double quotes are escaped. See
|
||||
# morituri.test.test_common_checksum
|
||||
|
||||
return path.replace('"', '\\"')
|
||||
|
||||
class Persister(object):
|
||||
"""
|
||||
I wrap an optional pickle to persist an object to disk.
|
||||
|
||||
@@ -158,9 +158,9 @@ class EncodeTask(task.Task):
|
||||
level name=level !
|
||||
%s !
|
||||
filesink location="%s" name=sink''' % (
|
||||
self._inpath.encode('utf-8'),
|
||||
common.quoteParse(self._inpath).encode('utf-8'),
|
||||
self._profile.pipeline,
|
||||
self._outpath.encode('utf-8')))
|
||||
common.quoteParse(self._outpath).encode('utf-8')))
|
||||
|
||||
tagger = self._pipeline.get_by_name('tagger')
|
||||
|
||||
|
||||
@@ -158,7 +158,8 @@ class AudioLengthTask(task.Task):
|
||||
self._pipeline = gst.parse_launch('''
|
||||
filesrc location="%s" !
|
||||
decodebin ! audio/x-raw-int !
|
||||
fakesink name=sink''' % self._path.encode('utf-8'))
|
||||
fakesink name=sink''' %
|
||||
common.quoteParse(self._path).encode('utf-8'))
|
||||
self.debug('pausing')
|
||||
self._pipeline.set_state(gst.STATE_PAUSED)
|
||||
self._pipeline.get_state()
|
||||
@@ -171,7 +172,9 @@ class AudioLengthTask(task.Task):
|
||||
try:
|
||||
length, qformat = sink.query_duration(gst.FORMAT_DEFAULT)
|
||||
except gst.QueryError:
|
||||
print 'failed to query %r' % self._path
|
||||
self.info('failed to query duration of %r' % self._path)
|
||||
raise
|
||||
|
||||
# wavparse 0.10.14 returns in bytes
|
||||
if qformat == gst.FORMAT_BYTES:
|
||||
self.debug('query returned in BYTES format')
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
# -*- Mode: Python -*-
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from morituri.common import log
|
||||
|
||||
log.init()
|
||||
@@ -30,3 +33,23 @@ def diffStrings(orig, new, desc='input'):
|
||||
return _diff(_tolines(orig),
|
||||
_tolines(new),
|
||||
desc=desc)
|
||||
|
||||
class TestCase(unittest.TestCase):
|
||||
# unittest.TestCase.failUnlessRaises does not return the exception,
|
||||
# and we'd like to check for the actual exception under TaskException,
|
||||
# so override the way twisted.trial.unittest does, without failure
|
||||
def failUnlessRaises(self, exception, f, *args, **kwargs):
|
||||
try:
|
||||
result = f(*args, **kwargs)
|
||||
except exception, inst:
|
||||
return inst
|
||||
except exception, e:
|
||||
raise self.failureException('%s raised instead of %s:\n %s'
|
||||
% (sys.exc_info()[0],
|
||||
exception.__name__,
|
||||
log.getExceptionMessage(e)))
|
||||
else:
|
||||
raise self.failureException('%s not raised (%r returned)'
|
||||
% (exception.__name__, result))
|
||||
|
||||
assertRaises = failUnlessRaises
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import os
|
||||
import unittest
|
||||
import tempfile
|
||||
|
||||
import gobject
|
||||
@@ -10,30 +9,45 @@ gobject.threads_init()
|
||||
|
||||
import gst
|
||||
|
||||
from morituri.common import task, checksum
|
||||
from morituri.test import common
|
||||
|
||||
from morituri.common import task, checksum, log
|
||||
|
||||
from morituri.test import common
|
||||
|
||||
def h(i):
|
||||
return "0x%08x" % i
|
||||
|
||||
class EmptyTestCase(unittest.TestCase):
|
||||
class EmptyTestCase(common.TestCase):
|
||||
def testEmpty(self):
|
||||
# this test makes sure that checksumming empty files doesn't hang
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp(suffix=u'morituri.test.empty')
|
||||
checksumtask = checksum.ChecksumTask(path)
|
||||
# FIXME: do we want a specific error for this ?
|
||||
self.assertRaises(gst.QueryError, self.runner.run,
|
||||
e = self.assertRaises(task.TaskException, self.runner.run,
|
||||
checksumtask, verbose=False)
|
||||
self.failUnless(isinstance(e.exception, gst.QueryError))
|
||||
os.unlink(path)
|
||||
|
||||
class PathTestCase(common.TestCase):
|
||||
def _testSuffix(self, suffix):
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp(suffix=suffix)
|
||||
checksumtask = checksum.ChecksumTask(path)
|
||||
e = self.assertRaises(task.TaskException, self.runner.run,
|
||||
checksumtask, verbose=False)
|
||||
self.failUnless(isinstance(e.exception, gst.QueryError))
|
||||
os.unlink(path)
|
||||
|
||||
def testUnicodePath(self):
|
||||
# this test makes sure we can checksum a unicode path
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp(
|
||||
suffix=u'morituri.test.B\xeate Noire.empty')
|
||||
checksumtask = checksum.ChecksumTask(path)
|
||||
self.assertRaises(gst.QueryError, self.runner.run,
|
||||
checksumtask, verbose=False)
|
||||
os.unlink(path)
|
||||
self._testSuffix(u'morituri.test.B\xeate Noire.empty')
|
||||
|
||||
def testSingleQuote(self):
|
||||
self._testSuffix(u"morituri.test.Guns 'N Roses")
|
||||
|
||||
def testDoubleQuote(self):
|
||||
# This test makes sure we can checksum files with double quote in
|
||||
# their name
|
||||
self._testSuffix(u'morituri.test.12" edit')
|
||||
|
||||
38
morituri/test/test_common_encode.py
Normal file
38
morituri/test/test_common_encode.py
Normal file
@@ -0,0 +1,38 @@
|
||||
# -*- Mode: Python; test-case-name: morituri.test.test_common_encode -*-
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import gobject
|
||||
gobject.threads_init()
|
||||
|
||||
import gst
|
||||
|
||||
from morituri.test import common
|
||||
|
||||
from morituri.common import task, encode, log
|
||||
|
||||
from morituri.test import common
|
||||
|
||||
class PathTestCase(common.TestCase):
|
||||
def _testSuffix(self, suffix):
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp(
|
||||
suffix=suffix)
|
||||
encodetask = encode.EncodeTask(path, path + '.out',
|
||||
encode.WavProfile())
|
||||
e = self.assertRaises(task.TaskException, self.runner.run,
|
||||
encodetask, verbose=False)
|
||||
self.failUnless(isinstance(e.exception, gst.QueryError))
|
||||
os.unlink(path)
|
||||
|
||||
def testUnicodePath(self):
|
||||
# this test makes sure we can checksum a unicode path
|
||||
self._testSuffix(u'morituri.test.B\xeate Noire.empty')
|
||||
|
||||
def testSingleQuote(self):
|
||||
self._testSuffix(u"morituri.test.Guns 'N Roses")
|
||||
|
||||
def testDoubleQuote(self):
|
||||
self._testSuffix(u'morituri.test.12" edit')
|
||||
@@ -2,13 +2,18 @@
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import gobject
|
||||
gobject.threads_init()
|
||||
|
||||
import gst
|
||||
|
||||
from morituri.image import image
|
||||
from morituri.common import task, common, log
|
||||
from morituri.test import common as tcommon
|
||||
|
||||
log.init()
|
||||
|
||||
def h(i):
|
||||
@@ -81,3 +86,26 @@ class AudioLengthTestCase(unittest.TestCase):
|
||||
runner = task.SyncRunner()
|
||||
runner.run(t, verbose=False)
|
||||
self.assertEquals(t.length, 10 * common.SAMPLES_PER_FRAME)
|
||||
|
||||
class AudioLengthPathTestCase(tcommon.TestCase):
|
||||
def _testSuffix(self, suffix):
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp(suffix=suffix)
|
||||
t = image.AudioLengthTask(path)
|
||||
e = self.assertRaises(task.TaskException, self.runner.run,
|
||||
t, verbose=False)
|
||||
self.failUnless(isinstance(e.exception, gst.QueryError),
|
||||
"%r is not a gst.QueryError" % e.exceptionMessage)
|
||||
os.unlink(path)
|
||||
|
||||
def testUnicodePath(self):
|
||||
# this test makes sure we can checksum a unicode path
|
||||
self._testSuffix(u'morituri.test.B\xeate Noire.empty')
|
||||
|
||||
def testSingleQuote(self):
|
||||
self._testSuffix(u"morituri.test.Guns 'N Roses")
|
||||
|
||||
def testDoubleQuote(self):
|
||||
# This test makes sure we can checksum files with double quote in
|
||||
# their name
|
||||
self._testSuffix(u'morituri.test.12" edit')
|
||||
|
||||
Reference in New Issue
Block a user