timeside/encoder/core.py: by default, refuse to overwrite an existing file, add corre...

diff --git a/tests/testencoding.py b/tests/testencoding.py
index 56ab7fa63a442c5ac8b9bf3d14d6d93e2f69987b..eac80498b46084e107dccceb0c2d8e2e1bcd4303 100644 (file)
--- 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 7718143c0ebb1deb0a011516fc7ba9ecaaaed8d4..e87587eb987df331bc587ffab2404a393a9e64e8 100644 (file)
--- 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 b0c65ecfccf2f4017b598350fb765984eea78f14..85dfd31fb6ebe4852047b2d8d2578ff285f4e5fa 100644 (file)
--- 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 3b1dd60a64b101036f29d4f4e37d261d4807e0f9..33d1f3791ed386963a304d688c296c46edcd49d4 100644 (file)
--- 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