Rip out all code that directly uses gstreamer
We can now rip CDs without gstreamer. This is not the most clean attempt, but I have tried to remove most of the code that depends on gstreamer. I hope there is not a lot of code left that depends on code that I have removed - I can at least rip a CD fully.
This commit is contained in:
@@ -1,67 +0,0 @@
|
||||
# -*- Mode: Python; test-case-name: morituri.test.test_common_checksum -*-
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import gobject
|
||||
gobject.threads_init()
|
||||
|
||||
from morituri.common import checksum, task as ctask
|
||||
|
||||
from morituri.extern.task import task, gstreamer
|
||||
|
||||
from morituri.test import common as tcommon
|
||||
|
||||
|
||||
def h(i):
|
||||
return "0x%08x" % i
|
||||
|
||||
|
||||
class EmptyTestCase(tcommon.TestCase):
|
||||
|
||||
def testEmpty(self):
|
||||
# this test makes sure that checksumming empty files doesn't hang
|
||||
self.runner = ctask.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 ?
|
||||
e = self.assertRaises(task.TaskException, self.runner.run,
|
||||
checksumtask, verbose=False)
|
||||
self.failUnless(isinstance(e.exception, gstreamer.GstException))
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
class PathTestCase(tcommon.TestCase):
|
||||
|
||||
def _testSuffix(self, suffix):
|
||||
self.runner = ctask.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, gstreamer.GstException))
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
class UnicodePathTestCase(PathTestCase, tcommon.UnicodeTestMixin):
|
||||
|
||||
def testUnicodePath(self):
|
||||
# this test makes sure we can checksum a unicode path
|
||||
self._testSuffix(u'morituri.test.B\xeate Noire.empty')
|
||||
|
||||
|
||||
class NormalPathTestCase(PathTestCase):
|
||||
|
||||
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')
|
||||
|
||||
def testBackSlash(self):
|
||||
# This test makes sure we can checksum files with a backslash in
|
||||
# their name
|
||||
self._testSuffix(u'morituri.test.40 Years Back\\Come')
|
||||
@@ -1,146 +0,0 @@
|
||||
# -*- 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.common import encode
|
||||
|
||||
from morituri.extern.task import task, gstreamer
|
||||
|
||||
from morituri.test import common
|
||||
|
||||
|
||||
class PathTestCase(common.TestCase):
|
||||
|
||||
def _testSuffix(self, suffix):
|
||||
# because of https://bugzilla.gnome.org/show_bug.cgi?id=688625
|
||||
# we first create the file with a 'normal' filename, then rename
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
fd, path = tempfile.mkstemp()
|
||||
|
||||
cmd = "gst-launch " \
|
||||
"audiotestsrc num-buffers=100 samplesperbuffer=1024 ! " \
|
||||
"audioconvert ! audio/x-raw-int,width=16,depth=16,channels =2 ! " \
|
||||
"wavenc ! " \
|
||||
"filesink location=\"%s\" > /dev/null 2>&1" % (
|
||||
gstreamer.quoteParse(path).encode('utf-8'), )
|
||||
self.debug('Running cmd %r' % cmd)
|
||||
os.system(cmd)
|
||||
self.failUnless(os.path.exists(path))
|
||||
os.close(fd)
|
||||
|
||||
fd, newpath = tempfile.mkstemp(suffix=suffix)
|
||||
os.rename(path, newpath)
|
||||
|
||||
encodetask = encode.EncodeTask(newpath, newpath + '.out',
|
||||
encode.WavProfile())
|
||||
self.runner.run(encodetask, verbose=False)
|
||||
os.close(fd)
|
||||
os.unlink(newpath)
|
||||
os.unlink(newpath + '.out')
|
||||
|
||||
|
||||
# class UnicodePathTestCase(PathTestCase, common.UnicodeTestMixin):
|
||||
|
||||
# def testUnicodePath(self):
|
||||
# # this test makes sure we can checksum a unicode path
|
||||
# self._testSuffix(u'.morituri.test_encode.B\xeate Noire')
|
||||
|
||||
|
||||
# class NormalPathTestCase(PathTestCase):
|
||||
|
||||
# def testSingleQuote(self):
|
||||
# self._testSuffix(u".morituri.test_encode.Guns 'N Roses")
|
||||
|
||||
# def testDoubleQuote(self):
|
||||
# self._testSuffix(u'.morituri.test_encode.12" edit')
|
||||
|
||||
|
||||
class TagReadTestCase(common.TestCase):
|
||||
|
||||
def testRead(self):
|
||||
path = os.path.join(os.path.dirname(__file__), u'track.flac')
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
t = encode.TagReadTask(path)
|
||||
self.runner.run(t)
|
||||
self.failUnless(t.taglist)
|
||||
self.assertEquals(t.taglist['audio-codec'], 'FLAC')
|
||||
self.assertEquals(t.taglist['description'], 'audiotest wave')
|
||||
|
||||
|
||||
# class TagWriteTestCase(common.TestCase):
|
||||
|
||||
# def testWrite(self):
|
||||
# fd, inpath = tempfile.mkstemp(suffix=u'.morituri.tagwrite.flac')
|
||||
|
||||
# # wave is pink-noise because a pure sine is encoded too efficiently
|
||||
# # by flacenc and triggers not enough frames in parsing
|
||||
# # FIXME: file a bug for this in GStreamer
|
||||
# os.system('gst-launch '
|
||||
# 'audiotestsrc '
|
||||
# 'wave=pink-noise num-buffers=10 samplesperbuffer=588 ! '
|
||||
# 'audioconvert ! '
|
||||
# 'audio/x-raw-int,channels=2,width=16,height=16,rate=44100 ! '
|
||||
# 'flacenc ! filesink location=%s > /dev/null 2>&1' % inpath)
|
||||
# os.close(fd)
|
||||
|
||||
# fd, outpath = tempfile.mkstemp(suffix=u'.morituri.tagwrite.flac')
|
||||
# self.runner = task.SyncRunner(verbose=False)
|
||||
# taglist = gst.TagList()
|
||||
# taglist[gst.TAG_ARTIST] = 'Artist'
|
||||
# taglist[gst.TAG_TITLE] = 'Title'
|
||||
|
||||
# t = encode.TagWriteTask(inpath, outpath, taglist)
|
||||
# self.runner.run(t)
|
||||
|
||||
# t = encode.TagReadTask(outpath)
|
||||
# self.runner.run(t)
|
||||
# self.failUnless(t.taglist)
|
||||
# self.assertEquals(t.taglist['audio-codec'], 'FLAC')
|
||||
# self.assertEquals(t.taglist['description'], 'audiotest wave')
|
||||
# self.assertEquals(t.taglist[gst.TAG_ARTIST], 'Artist')
|
||||
# self.assertEquals(t.taglist[gst.TAG_TITLE], 'Title')
|
||||
|
||||
# os.unlink(inpath)
|
||||
# os.unlink(outpath)
|
||||
|
||||
|
||||
class SafeRetagTestCase(common.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self._fd, self._path = tempfile.mkstemp(suffix=u'.morituri.retag.flac')
|
||||
|
||||
os.system('gst-launch '
|
||||
'audiotestsrc '
|
||||
'num-buffers=40 samplesperbuffer=588 wave=pink-noise ! '
|
||||
'audioconvert ! '
|
||||
'audio/x-raw-int,channels=2,width=16,height=16,rate=44100 ! '
|
||||
'flacenc ! filesink location=%s > /dev/null 2>&1' % self._path)
|
||||
os.close(self._fd)
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
|
||||
def tearDown(self):
|
||||
os.unlink(self._path)
|
||||
|
||||
# def testNoChange(self):
|
||||
# taglist = gst.TagList()
|
||||
# taglist[gst.TAG_DESCRIPTION] = 'audiotest wave'
|
||||
# taglist[gst.TAG_AUDIO_CODEC] = 'FLAC'
|
||||
|
||||
# t = encode.SafeRetagTask(self._path, taglist)
|
||||
# self.runner.run(t)
|
||||
|
||||
# def testChange(self):
|
||||
# taglist = gst.TagList()
|
||||
# taglist[gst.TAG_DESCRIPTION] = 'audiotest retagged'
|
||||
# taglist[gst.TAG_AUDIO_CODEC] = 'FLAC'
|
||||
# taglist[gst.TAG_ARTIST] = 'Artist'
|
||||
|
||||
# t = encode.SafeRetagTask(self._path, taglist)
|
||||
# self.runner.run(t)
|
||||
@@ -1,21 +0,0 @@
|
||||
# -*- Mode: Python -*-
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
from morituri.common import gstreamer
|
||||
|
||||
from morituri.test import common
|
||||
|
||||
|
||||
class VersionTestCase(common.TestCase):
|
||||
|
||||
def testGStreamer(self):
|
||||
version = gstreamer.gstreamerVersion()
|
||||
self.failUnless(version.startswith('0.'))
|
||||
|
||||
def testGSTPython(self):
|
||||
version = gstreamer.gstPythonVersion()
|
||||
self.failUnless(version.startswith('0.'))
|
||||
|
||||
def testFlacEnc(self):
|
||||
version = gstreamer.elementFactoryVersion('flacenc')
|
||||
self.failUnless(version.startswith('0.'))
|
||||
@@ -1,85 +0,0 @@
|
||||
# -*- Mode: Python; test-case-name: morituri.test.test_image_image -*-
|
||||
# vi:si:et:sw=4:sts=4:ts=4
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import gobject
|
||||
gobject.threads_init()
|
||||
|
||||
import gst
|
||||
|
||||
from morituri.image import image
|
||||
from morituri.common import common
|
||||
|
||||
from morituri.extern.task import task, gstreamer
|
||||
|
||||
from morituri.test import common as tcommon
|
||||
|
||||
|
||||
def h(i):
|
||||
return "0x%08x" % i
|
||||
|
||||
|
||||
class TrackSingleTestCase(tcommon.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.image = image.Image(os.path.join(os.path.dirname(__file__),
|
||||
u'track-single.cue'))
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
self.image.setup(self.runner)
|
||||
|
||||
def testAccurateRipChecksum(self):
|
||||
checksumtask = image.AccurateRipChecksumTask(self.image)
|
||||
self.runner.run(checksumtask, verbose=False)
|
||||
|
||||
self.assertEquals(len(checksumtask.checksums), 4)
|
||||
# self.assertEquals(h(checksumtask.checksums[0]), '0x00000000')
|
||||
# self.assertEquals(h(checksumtask.checksums[1]), '0x793fa868')
|
||||
# self.assertEquals(h(checksumtask.checksums[2]), '0x8dd37c26')
|
||||
# self.assertEquals(h(checksumtask.checksums[3]), '0x00000000')
|
||||
|
||||
def testLength(self):
|
||||
self.assertEquals(self.image.table.getTrackLength(1), 2)
|
||||
self.assertEquals(self.image.table.getTrackLength(2), 2)
|
||||
self.assertEquals(self.image.table.getTrackLength(3), 2)
|
||||
self.assertEquals(self.image.table.getTrackLength(4), 4)
|
||||
|
||||
def testCDDB(self):
|
||||
self.assertEquals(self.image.table.getCDDBDiscId(), "08000004")
|
||||
|
||||
def testAccurateRip(self):
|
||||
self.assertEquals(self.image.table.getAccurateRipIds(), (
|
||||
"00000016", "0000005b"))
|
||||
|
||||
|
||||
class TrackSeparateTestCase(tcommon.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.image = image.Image(os.path.join(os.path.dirname(__file__),
|
||||
u'track-separate.cue'))
|
||||
self.runner = task.SyncRunner(verbose=False)
|
||||
self.image.setup(self.runner)
|
||||
|
||||
def testAccurateRipChecksum(self):
|
||||
checksumtask = image.AccurateRipChecksumTask(self.image)
|
||||
self.runner.run(checksumtask, verbose=False)
|
||||
|
||||
self.assertEquals(len(checksumtask.checksums), 4)
|
||||
self.assertEquals(h(checksumtask.checksums[0]), '0xd60e55e1')
|
||||
self.assertEquals(h(checksumtask.checksums[1]), '0xd63dc2d2')
|
||||
self.assertEquals(h(checksumtask.checksums[2]), '0xd63dc2d2')
|
||||
self.assertEquals(h(checksumtask.checksums[3]), '0x7271db39')
|
||||
|
||||
def testLength(self):
|
||||
self.assertEquals(self.image.table.getTrackLength(1), 10)
|
||||
self.assertEquals(self.image.table.getTrackLength(2), 10)
|
||||
self.assertEquals(self.image.table.getTrackLength(3), 10)
|
||||
self.assertEquals(self.image.table.getTrackLength(4), 10)
|
||||
|
||||
def testCDDB(self):
|
||||
self.assertEquals(self.image.table.getCDDBDiscId(), "08000004")
|
||||
|
||||
def testAccurateRip(self):
|
||||
self.assertEquals(self.image.table.getAccurateRipIds(), (
|
||||
"00000064", "00000191"))
|
||||
Reference in New Issue
Block a user