From 25df40cbcfb179faab76acd8b27060352537a325 Mon Sep 17 00:00:00 2001 From: Paul Brossier Date: Sun, 27 Jan 2013 20:46:14 -0600 Subject: [PATCH] timeside/encoder/core.py: by default, refuse to overwrite an existing file, add corresponding tests --- tests/testencoding.py | 90 +++++++++++++++++++++++++++++++++++++++- timeside/encoder/core.py | 4 +- timeside/encoder/m4a.py | 4 +- timeside/encoder/webm.py | 4 +- 4 files changed, 96 insertions(+), 6 deletions(-) diff --git a/tests/testencoding.py b/tests/testencoding.py index 56ab7fa..eac8049 100644 --- a/tests/testencoding.py +++ b/tests/testencoding.py @@ -3,6 +3,7 @@ from numpy import arange, sin, zeros from unit_timeside import * import os.path +from os import unlink class TestEncoding(TestCase): "Test encoding features" @@ -12,6 +13,7 @@ class TestEncoding(TestCase): import tempfile self.tmpfile = tempfile.NamedTemporaryFile(delete=True) self.sink = self.tmpfile.name + self.tmpfile.close() def testWav(self): "Test wav encoding" @@ -74,7 +76,8 @@ class TestEncoding(TestCase): print commands.getoutput('sndfile-info ' + self.sink) self.assertEquals(written_frames, total_frames) - if hasattr(self, 'tmpfile'): self.tmpfile.close() + + if hasattr(self, 'tmpfile'): unlink(self.sink) class TestEncodingLongBlock(TestEncoding): "Test encoding features with longer blocksize" @@ -160,5 +163,90 @@ class TestEncodingToDirectory(TestEncoding): from os import rmdir rmdir(self.sink) +class TestEncodingOverwriteFails(TestCase): + "Test encoding features" + + def setUp(self): + self.samplerate, self.channels, self.blocksize = 44100, 1, 1024 + import tempfile + self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + self.sink = self.tmpfile.name + + def testWav(self): + "Test wav encoding" + from timeside.encoder.wav import WavEncoder + self.assertRaises(IOError,WavEncoder,self.sink) + + def testVorbis(self): + "Test vorbis encoding" + from timeside.encoder.ogg import VorbisEncoder + self.assertRaises(IOError,VorbisEncoder,self.sink) + + def testMp3(self): + "Test mp3 encoding" + from timeside.encoder.mp3 import Mp3Encoder + self.assertRaises(IOError,Mp3Encoder,self.sink) + + def testAac(self): + "Test aac encoding" + from timeside.encoder.m4a import AacEncoder + self.assertRaises(IOError,AacEncoder,self.sink) + + def testFlac(self): + "Test flac encoding" + from timeside.encoder.flac import FlacEncoder + self.assertRaises(IOError,FlacEncoder,self.sink) + + def testWebm(self): + "Test webm encoding" + from timeside.encoder.webm import WebMEncoder + self.assertRaises(IOError,WebMEncoder,self.sink) + + def tearDown(self): + self.tmpfile.close() + +class TestEncodingOverwriteForced(TestCase): + "Test encoding features" + + def setUp(self): + self.samplerate, self.channels, self.blocksize = 44100, 1, 1024 + import tempfile + self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + self.sink = self.tmpfile.name + + def testWav(self): + "Test wav encoding" + from timeside.encoder.wav import WavEncoder + self.encoder = WavEncoder(self.sink, overwrite = True) + + def testVorbis(self): + "Test vorbis encoding" + from timeside.encoder.ogg import VorbisEncoder + self.encoder = VorbisEncoder(self.sink, overwrite = True) + + def testMp3(self): + "Test mp3 encoding" + from timeside.encoder.mp3 import Mp3Encoder + self.encoder = Mp3Encoder(self.sink, overwrite = True) + + def testAac(self): + "Test aac encoding" + from timeside.encoder.m4a import AacEncoder + self.encoder = AacEncoder(self.sink, overwrite = True) + + def testFlac(self): + "Test flac encoding" + from timeside.encoder.flac import FlacEncoder + self.encoder = FlacEncoder(self.sink, overwrite = True) + + def testWebm(self): + "Test webm encoding" + from timeside.encoder.webm import WebMEncoder + self.encoder = WebMEncoder(self.sink, overwrite = True) + + def tearDown(self): + super(TestEncodingOverwriteForced, self).tearDown() + self.tmpfile.close() + if __name__ == '__main__': unittest.main(testRunner=TestRunner()) diff --git a/timeside/encoder/core.py b/timeside/encoder/core.py index 7718143..e87587e 100644 --- a/timeside/encoder/core.py +++ b/timeside/encoder/core.py @@ -26,11 +26,13 @@ from timeside.tools import * class GstEncoder(Processor): - def __init__(self, output, streaming = False): + def __init__(self, output, streaming = False, overwrite = False): if isinstance(output, basestring): import os.path if os.path.isdir(output): raise IOError("Encoder output must be a file, not a directory") + elif os.path.isfile(output) and not overwrite: + raise IOError("Encoder output %s exists, but overwrite set to False") self.filename = output else: self.filename = None diff --git a/timeside/encoder/m4a.py b/timeside/encoder/m4a.py index b0c65ec..85dfd31 100644 --- a/timeside/encoder/m4a.py +++ b/timeside/encoder/m4a.py @@ -29,8 +29,8 @@ class AacEncoder(GstEncoder): """ gstreamer-based AAC encoder """ implements(IEncoder) - def __init__(self, output, streaming = False): - super(AacEncoder, self).__init__(output, streaming) + def __init__(self, *args, **kwargs): + super(AacEncoder, self).__init__( *args, **kwargs) if self.streaming: raise Exception("Streaming not supported") diff --git a/timeside/encoder/webm.py b/timeside/encoder/webm.py index 3b1dd60..33d1f37 100644 --- a/timeside/encoder/webm.py +++ b/timeside/encoder/webm.py @@ -29,8 +29,8 @@ class WebMEncoder(GstEncoder): """ gstreamer-based webm encoder and muxer """ implements(IEncoder) - def __init__(self, output, streaming = False): - super(WebMEncoder, self).__init__(output, streaming) + def __init__(self, *args, **kwargs): + super(WebMEncoder, self).__init__(*args, **kwargs) self.video = False @interfacedoc -- 2.39.5