Ignore:
Timestamp:
Feb 23, 2014, 6:34:34 PM (11 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

File:
1 edited

Legend:

Unmodified
Added
Removed
  • python/tests/test_sink.py

    r18d8bef rab35262  
    44from aubio import fvec, source, sink
    55from numpy import array
    6 from utils import list_all_sounds
     6from utils import list_all_sounds, get_tmp_sink_path, del_tmp_sink_path
    77
    88list_of_sounds = list_all_sounds('sounds')
    99path = None
    1010
     11many_files = 300 # 256 opened files is too much
     12
    1113class aubio_sink_test_case(TestCase):
    1214
    1315    def setUp(self):
    14         if not len(list_of_sounds): self.skipTest('add some sound files in \'python/tests/sounds\'')
     16        if not len(list_of_sounds):
     17            self.skipTest('add some sound files in \'python/tests/sounds\'')
    1518
    1619    def test_many_sinks(self):
    17         for i in range(100):
    18             g = sink('/tmp/f.wav', 0)
    19             write = 256
     20        from tempfile import mkdtemp
     21        import os.path
     22        import shutil
     23        tmpdir = mkdtemp()
     24        sink_list = []
     25        for i in range(many_files):
     26            path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
     27            g = sink(path, 0)
     28            sink_list.append(g)
     29            write = 32
    2030            for n in range(200):
    2131                vec = fvec(write)
    2232                g(vec, write)
    23             del g
     33            g.close()
     34        shutil.rmtree(tmpdir)
     35
     36    def test_many_sinks_not_closed(self):
     37        from tempfile import mkdtemp
     38        import os.path
     39        import shutil
     40        tmpdir = mkdtemp()
     41        sink_list = []
     42        try:
     43            for i in range(many_files):
     44                path = os.path.join(tmpdir, 'f-' + str(i) + '.wav')
     45                g = sink(path, 0)
     46                sink_list.append(g)
     47                write = 256
     48                for n in range(200):
     49                    vec = fvec(write)
     50                    g(vec, write)
     51        except StandardError:
     52            pass
     53        else:
     54            self.fail("does not fail on too many files open")
     55        for g in sink_list:
     56            g.close()
     57        shutil.rmtree(tmpdir)
    2458
    2559    def test_read(self):
     
    2862                f = source(path, samplerate, hop_size)
    2963                if samplerate == 0: samplerate = f.samplerate
    30                 g = sink('/tmp/f.wav', samplerate)
     64                sink_path = get_tmp_sink_path()
     65                g = sink(sink_path, samplerate)
    3166                total_frames = 0
    3267                while True:
     
    3570                    total_frames += read
    3671                    if read < f.hop_size: break
    37                 print "read", "%.2fs" % (total_frames / float(f.samplerate) ),
    38                 print "(", total_frames, "frames", "in",
    39                 print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")",
    40                 print "from", f.uri,
    41                 print "to", g.uri
    42                 #del f, g
     72                if 0:
     73                    print "read", "%.2fs" % (total_frames / float(f.samplerate) ),
     74                    print "(", total_frames, "frames", "in",
     75                    print total_frames / f.hop_size, "blocks", "at", "%dHz" % f.samplerate, ")",
     76                    print "from", f.uri,
     77                    print "to", g.uri
     78                del_tmp_sink_path(sink_path)
    4379
    4480    def test_close_file(self):
    4581        samplerate = 44100
    46         g = sink('/tmp/f.wav', samplerate)
     82        sink_path = get_tmp_sink_path()
     83        g = sink(sink_path, samplerate)
    4784        g.close()
     85        del_tmp_sink_path(sink_path)
    4886
    4987    def test_close_file_twice(self):
    5088        samplerate = 44100
    51         g = sink('/tmp/f.wav', samplerate)
     89        sink_path = get_tmp_sink_path()
     90        g = sink(sink_path, samplerate)
    5291        g.close()
    5392        g.close()
     93        del_tmp_sink_path(sink_path)
    5494
    5595if __name__ == '__main__':
Note: See TracChangeset for help on using the changeset viewer.