From ab352628702f0b2d69284f3e18e1a6cd53011122 Mon Sep 17 00:00:00 2001
From: Paul Brossier <piem@piem.org>
Date: Sun, 23 Feb 2014 14:34:34 -0300
Subject: [PATCH] python/tests/test_sink.py: add more tests, clean up created
 files

---
 python/tests/test_sink.py | 70 +++++++++++++++++++++++++++++++++++++----------
 python/tests/utils.py     | 10 +++++--
 2 files changed, 62 insertions(+), 18 deletions(-)

diff --git a/python/tests/test_sink.py b/python/tests/test_sink.py
index ffa83845..5c8aa8f7 100755
--- a/python/tests/test_sink.py
+++ b/python/tests/test_sink.py
@@ -3,54 +3,94 @@
 from numpy.testing import TestCase, assert_equal, assert_almost_equal
 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):
         for path in list_of_sounds:
             for samplerate, hop_size in zip([0, 44100, 8000, 32000], [512, 1024, 64, 256]):
                 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:
                     vec, read = f()
                     g(vec, read)
                     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__':
     from unittest import main
diff --git a/python/tests/utils.py b/python/tests/utils.py
index 8212cbdd..1cec0530 100644
--- a/python/tests/utils.py
+++ b/python/tests/utils.py
@@ -19,12 +19,16 @@ def get_default_test_sound(TestCase, rel_dir = 'sounds'):
     else:
         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):
     import yaml
-- 
2.11.0