Changeset ab35262 for python/tests

Timestamp:

Feb 23, 2014, 6:34:34 PM (10 years ago)

Author:

Paul Brossier <piem@piem.org>

Branches:

feature/autosink, feature/cnn, feature/cnn_org, feature/constantq, feature/crepe, feature/crepe_org, feature/pitchshift, feature/pydocstrings, feature/timestretch, fix/ffmpeg5, master, pitchshift, sampler, timestretch, yinfft+

Children:

4a1378c

Parents:

18d8bef

Message:

python/tests/test_sink.py: add more tests, clean up created files

Location:

python/tests

Files:

• test_sink.py (modified) (3 diffs)
• utils.py (modified) (1 diff)

python/tests/test_sink.py

@@ -4 +4 @@
 from aubio import fvec, source, sink
 from numpy import array
-from utils import list_all_sounds
+from utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path
 
 list_of_sounds = list_all_sounds('sounds')
 path = None
 
+many_files = 300 # 256 opened files is too much
+
 class aubio_sink_test_case(TestCase):
 
     def setUp(self):
-        if not len(list_of_sounds): self.skipTest('add some sound files in \'python/tests/sounds\'')
+        if not len(list_of_sounds):
+            self.skipTest('add some sound files in \'python/tests/sounds\'')
 
     def test_many_sinks(self):
-        for i in range(100):
-            g = sink('/tmp/f.wav', 0)
-            write = 256
+        from tempfile import mkdtemp
+        import os.path
+        import shutil
+        tmpdir = mkdtemp()
+        sink_list = []
+        for i in range(many_files):
+            path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
+            g = sink(path, 0)
+            sink_list.append(g)
+            write = 32
             for n in range(200):
                 vec = fvec(write)
                 g(vec, write)
-            del g
+            g.close()
+        shutil.rmtree(tmpdir)
+
+    def test_many_sinks_not_closed(self):
+        from tempfile import mkdtemp
+        import os.path
+        import shutil
+        tmpdir = mkdtemp()
+        sink_list = []
+        try:
+            for i in range(many_files):
+                path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
+                g = sink(path, 0)
+                sink_list.append(g)
+                write = 256
+                for n in range(200):
+                    vec = fvec(write)
+                    g(vec, write)
+        except StandardError:
+            pass
+        else:
+            self.fail("does not fail on too many files open")
+        for g in sink_list:
+            g.close()
+        shutil.rmtree(tmpdir)
 
     def test_read(self):
@@ -28 +62 @@
                 f = source(path, samplerate, hop_size)
                 if samplerate == 0: samplerate = f.samplerate
-                g = sink('/tmp/f.wav', samplerate)
+                sink_path = get_tmp_sink_path()
+                g = sink(sink_path, samplerate)
                 total_frames = 0
                 while True:
@@ -35 +70 @@
                     total_frames += read
                     if read < f.hop_size: break
-                print "read", "%.2fs" % (total_frames / float(f.samplerate) ),
-                print "(", total_frames, "frames", "in",
-                print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")",
-                print "from", f.uri,
-                print "to", g.uri
-                #del f, g
+                if 0:
+                    print "read", "%.2fs" % (total_frames / float(f.samplerate) ),
+                    print "(", total_frames, "frames", "in",
+                    print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")",
+                    print "from", f.uri,
+                    print "to", g.uri
+                del_tmp_sink_path(sink_path)
 
     def test_close_file(self):
         samplerate = 44100
-        g = sink('/tmp/f.wav', samplerate)
+        sink_path = get_tmp_sink_path()
+        g = sink(sink_path, samplerate)
         g.close()
+        del_tmp_sink_path(sink_path)
 
     def test_close_file_twice(self):
         samplerate = 44100
-        g = sink('/tmp/f.wav', samplerate)
+        sink_path = get_tmp_sink_path()
+        g = sink(sink_path, samplerate)
         g.close()
         g.close()
+        del_tmp_sink_path(sink_path)
 
 if __name__ == '__main__':

python/tests/utils.py

@@ -20 +20 @@
         return all_sounds[0]
 
-def get_tmp_sink_name():
+def get_tmp_sink_path():
     from tempfile import mkstemp
     import os
-    fd, name = mkstemp()
+    fd, path = mkstemp()
     os.close(fd)
-    return name
+    return path
+
+def del_tmp_sink_path(path):
+    import os
+    os.unlink(path)
 
 def array_from_yaml_file(filename):