--- /dev/null
- f = self.module.open(testfile, 'rb')
- try:
+ from test.support import findfile, TESTFN, unlink
+ import unittest
+ import array
+ import io
+ import pickle
+ import sys
+
+ def byteswap2(data):
+ a = array.array('h', data)
+ a.byteswap()
+ return a.tobytes()
+
+ def byteswap3(data):
+ ba = bytearray(data)
+ ba[::3] = data[2::3]
+ ba[2::3] = data[::3]
+ return bytes(ba)
+
+ def byteswap4(data):
+ a = array.array('i', data)
+ a.byteswap()
+ return a.tobytes()
+
+
+ class AudioTests:
+ close_fd = False
+
+ def setUp(self):
+ self.f = self.fout = None
+
+ def tearDown(self):
+ if self.f is not None:
+ self.f.close()
+ if self.fout is not None:
+ self.fout.close()
+ unlink(TESTFN)
+
+ def check_params(self, f, nchannels, sampwidth, framerate, nframes,
+ comptype, compname):
+ self.assertEqual(f.getnchannels(), nchannels)
+ self.assertEqual(f.getsampwidth(), sampwidth)
+ self.assertEqual(f.getframerate(), framerate)
+ self.assertEqual(f.getnframes(), nframes)
+ self.assertEqual(f.getcomptype(), comptype)
+ self.assertEqual(f.getcompname(), compname)
+
+ params = f.getparams()
+ self.assertEqual(params,
+ (nchannels, sampwidth, framerate, nframes, comptype, compname))
++ self.assertEqual(params.nchannels, nchannels)
++ self.assertEqual(params.sampwidth, sampwidth)
++ self.assertEqual(params.framerate, framerate)
++ self.assertEqual(params.nframes, nframes)
++ self.assertEqual(params.comptype, comptype)
++ self.assertEqual(params.compname, compname)
+
+ dump = pickle.dumps(params)
+ self.assertEqual(pickle.loads(dump), params)
+
+
+ class AudioWriteTests(AudioTests):
+
+ def create_file(self, testfile):
+ f = self.fout = self.module.open(testfile, 'wb')
+ f.setnchannels(self.nchannels)
+ f.setsampwidth(self.sampwidth)
+ f.setframerate(self.framerate)
+ f.setcomptype(self.comptype, self.compname)
+ return f
+
+ def check_file(self, testfile, nframes, frames):
- finally:
- f.close()
++ with self.module.open(testfile, 'rb') as f:
+ self.assertEqual(f.getnchannels(), self.nchannels)
+ self.assertEqual(f.getsampwidth(), self.sampwidth)
+ self.assertEqual(f.getframerate(), self.framerate)
+ self.assertEqual(f.getnframes(), nframes)
+ self.assertEqual(f.readframes(nframes), frames)
- f = self.module.open(testfile, 'rb')
- try:
+
+ def test_write_params(self):
+ f = self.create_file(TESTFN)
+ f.setnframes(self.nframes)
+ f.writeframes(self.frames)
+ self.check_params(f, self.nchannels, self.sampwidth, self.framerate,
+ self.nframes, self.comptype, self.compname)
+ f.close()
+
++ def test_write_context_manager_calls_close(self):
++ # Close checks for a minimum header and will raise an error
++ # if it is not set, so this proves that close is called.
++ with self.assertRaises(self.module.Error):
++ with self.module.open(TESTFN, 'wb'):
++ pass
++ with self.assertRaises(self.module.Error):
++ with open(TESTFN, 'wb') as testfile:
++ with self.module.open(testfile):
++ pass
++
++ def test_context_manager_with_open_file(self):
++ with open(TESTFN, 'wb') as testfile:
++ with self.module.open(testfile) as f:
++ f.setnchannels(self.nchannels)
++ f.setsampwidth(self.sampwidth)
++ f.setframerate(self.framerate)
++ f.setcomptype(self.comptype, self.compname)
++ self.assertEqual(testfile.closed, self.close_fd)
++ with open(TESTFN, 'rb') as testfile:
++ with self.module.open(testfile) as f:
++ self.assertFalse(f.getfp().closed)
++ params = f.getparams()
++ self.assertEqual(params.nchannels, self.nchannels)
++ self.assertEqual(params.sampwidth, self.sampwidth)
++ self.assertEqual(params.framerate, self.framerate)
++ if not self.close_fd:
++ self.assertIsNone(f.getfp())
++ self.assertEqual(testfile.closed, self.close_fd)
++
++ def test_context_manager_with_filename(self):
++ # If the file doesn't get closed, this test won't fail, but it will
++ # produce a resource leak warning.
++ with self.module.open(TESTFN, 'wb') as f:
++ f.setnchannels(self.nchannels)
++ f.setsampwidth(self.sampwidth)
++ f.setframerate(self.framerate)
++ f.setcomptype(self.comptype, self.compname)
++ with self.module.open(TESTFN) as f:
++ self.assertFalse(f.getfp().closed)
++ params = f.getparams()
++ self.assertEqual(params.nchannels, self.nchannels)
++ self.assertEqual(params.sampwidth, self.sampwidth)
++ self.assertEqual(params.framerate, self.framerate)
++ if not self.close_fd:
++ self.assertIsNone(f.getfp())
++
+ def test_write(self):
+ f = self.create_file(TESTFN)
+ f.setnframes(self.nframes)
+ f.writeframes(self.frames)
+ f.close()
+
+ self.check_file(TESTFN, self.nframes, self.frames)
+
+ def test_incompleted_write(self):
+ with open(TESTFN, 'wb') as testfile:
+ testfile.write(b'ababagalamaga')
+ f = self.create_file(testfile)
+ f.setnframes(self.nframes + 1)
+ f.writeframes(self.frames)
+ f.close()
+
+ with open(TESTFN, 'rb') as testfile:
+ self.assertEqual(testfile.read(13), b'ababagalamaga')
+ self.check_file(testfile, self.nframes, self.frames)
+
+ def test_multiple_writes(self):
+ with open(TESTFN, 'wb') as testfile:
+ testfile.write(b'ababagalamaga')
+ f = self.create_file(testfile)
+ f.setnframes(self.nframes)
+ framesize = self.nchannels * self.sampwidth
+ f.writeframes(self.frames[:-framesize])
+ f.writeframes(self.frames[-framesize:])
+ f.close()
+
+ with open(TESTFN, 'rb') as testfile:
+ self.assertEqual(testfile.read(13), b'ababagalamaga')
+ self.check_file(testfile, self.nframes, self.frames)
+
+ def test_overflowed_write(self):
+ with open(TESTFN, 'wb') as testfile:
+ testfile.write(b'ababagalamaga')
+ f = self.create_file(testfile)
+ f.setnframes(self.nframes - 1)
+ f.writeframes(self.frames)
+ f.close()
+
+ with open(TESTFN, 'rb') as testfile:
+ self.assertEqual(testfile.read(13), b'ababagalamaga')
+ self.check_file(testfile, self.nframes, self.frames)
+
+
+ class AudioTestsWithSourceFile(AudioTests):
+
+ @classmethod
+ def setUpClass(cls):
+ cls.sndfilepath = findfile(cls.sndfilename, subdir='audiodata')
+
+ def test_read_params(self):
+ f = self.f = self.module.open(self.sndfilepath)
+ #self.assertEqual(f.getfp().name, self.sndfilepath)
+ self.check_params(f, self.nchannels, self.sampwidth, self.framerate,
+ self.sndfilenframes, self.comptype, self.compname)
+
+ def test_close(self):
+ testfile = open(self.sndfilepath, 'rb')
+ f = self.f = self.module.open(testfile)
+ self.assertFalse(testfile.closed)
+ f.close()
+ self.assertEqual(testfile.closed, self.close_fd)
+ testfile = open(TESTFN, 'wb')
+ fout = self.module.open(testfile, 'wb')
+ self.assertFalse(testfile.closed)
+ with self.assertRaises(self.module.Error):
+ fout.close()
+ self.assertEqual(testfile.closed, self.close_fd)
+ fout.close() # do nothing
+
+ def test_read(self):
+ framesize = self.nchannels * self.sampwidth
+ chunk1 = self.frames[:2 * framesize]
+ chunk2 = self.frames[2 * framesize: 4 * framesize]
+ f = self.f = self.module.open(self.sndfilepath)
+ self.assertEqual(f.readframes(0), b'')
+ self.assertEqual(f.tell(), 0)
+ self.assertEqual(f.readframes(2), chunk1)
+ f.rewind()
+ pos0 = f.tell()
+ self.assertEqual(pos0, 0)
+ self.assertEqual(f.readframes(2), chunk1)
+ pos2 = f.tell()
+ self.assertEqual(pos2, 2)
+ self.assertEqual(f.readframes(2), chunk2)
+ f.setpos(pos2)
+ self.assertEqual(f.readframes(2), chunk2)
+ f.setpos(pos0)
+ self.assertEqual(f.readframes(2), chunk1)
+ with self.assertRaises(self.module.Error):
+ f.setpos(-1)
+ with self.assertRaises(self.module.Error):
+ f.setpos(f.getnframes() + 1)
+
+ def test_copy(self):
+ f = self.f = self.module.open(self.sndfilepath)
+ fout = self.fout = self.module.open(TESTFN, 'wb')
+ fout.setparams(f.getparams())
+ i = 0
+ n = f.getnframes()
+ while n > 0:
+ i += 1
+ fout.writeframes(f.readframes(i))
+ n -= i
+ fout.close()
+ fout = self.fout = self.module.open(TESTFN, 'rb')
+ f.rewind()
+ self.assertEqual(f.getparams(), fout.getparams())
+ self.assertEqual(f.readframes(f.getnframes()),
+ fout.readframes(fout.getnframes()))
+
+ def test_read_not_from_start(self):
+ with open(TESTFN, 'wb') as testfile:
+ testfile.write(b'ababagalamaga')
+ with open(self.sndfilepath, 'rb') as f:
+ testfile.write(f.read())
+
+ with open(TESTFN, 'rb') as testfile:
+ self.assertEqual(testfile.read(13), b'ababagalamaga')
- finally:
- f.close()
++ with self.module.open(testfile, 'rb') as f:
+ self.assertEqual(f.getnchannels(), self.nchannels)
+ self.assertEqual(f.getsampwidth(), self.sampwidth)
+ self.assertEqual(f.getframerate(), self.framerate)
+ self.assertEqual(f.getnframes(), self.sndfilenframes)
+ self.assertEqual(f.readframes(self.nframes), self.frames)
def test_skipunknown(self):
#Issue 2245
#This file contains chunk types aifc doesn't recognize.
- self.f = aifc.open(self.sndfilepath)
-
- def test_params(self):
- f = self.f = aifc.open(self.sndfilepath)
- params = f.getparams()
- self.assertEqual(f.getfp().name, self.sndfilepath)
- self.assertEqual(f.getnchannels(), 2)
- self.assertEqual(f.getsampwidth(), 2)
- self.assertEqual(f.getframerate(), 48000)
- self.assertEqual(f.getnframes(), 14400)
- self.assertEqual(f.getcomptype(), b'NONE')
- self.assertEqual(f.getcompname(), b'not compressed')
- self.assertEqual(
- f.getparams(),
- (2, 2, 48000, 14400, b'NONE', b'not compressed'),
- )
-
- params = f.getparams()
- self.assertEqual(params.nchannels, 2)
- self.assertEqual(params.sampwidth, 2)
- self.assertEqual(params.framerate, 48000)
- self.assertEqual(params.nframes, 14400)
- self.assertEqual(params.comptype, b'NONE')
- self.assertEqual(params.compname, b'not compressed')
+ self.f = aifc.open(findfile('Sine-1000Hz-300ms.aif'))
- def test_getparams_picklable(self):
- self.f = aifc.open(self.sndfilepath)
- params = self.f.getparams()
- dump = pickle.dumps(params)
- self.assertEqual(pickle.loads(dump), params)
- self.f.close()
-
- def test_context_manager(self):
- with open(self.sndfilepath, 'rb') as testfile:
- with aifc.open(testfile) as f:
- pass
- self.assertEqual(testfile.closed, True)
- with open(TESTFN, 'wb') as testfile:
- with self.assertRaises(aifc.Error):
- with aifc.open(testfile, 'wb') as fout:
- pass
- self.assertEqual(testfile.closed, True)
- fout.close() # do nothing
-
- def test_read(self):
- f = self.f = aifc.open(self.sndfilepath)
- self.assertEqual(f.readframes(0), b'')
- self.assertEqual(f.tell(), 0)
- self.assertEqual(f.readframes(2), b'\x00\x00\x00\x00\x0b\xd4\x0b\xd4')
- f.rewind()
- pos0 = f.tell()
- self.assertEqual(pos0, 0)
- self.assertEqual(f.readframes(2), b'\x00\x00\x00\x00\x0b\xd4\x0b\xd4')
- pos2 = f.tell()
- self.assertEqual(pos2, 2)
- self.assertEqual(f.readframes(2), b'\x17t\x17t"\xad"\xad')
- f.setpos(pos2)
- self.assertEqual(f.readframes(2), b'\x17t\x17t"\xad"\xad')
- f.setpos(pos0)
- self.assertEqual(f.readframes(2), b'\x00\x00\x00\x00\x0b\xd4\x0b\xd4')
- with self.assertRaises(aifc.Error):
- f.setpos(-1)
- with self.assertRaises(aifc.Error):
- f.setpos(f.getnframes() + 1)
-
- def test_write(self):
- f = self.f = aifc.open(self.sndfilepath)
- fout = self.fout = aifc.open(TESTFN, 'wb')
- fout.aifc()
- fout.setparams(f.getparams())
- for frame in range(f.getnframes()):
- fout.writeframes(f.readframes(1))
- fout.close()
- fout = self.fout = aifc.open(TESTFN, 'rb')
- f.rewind()
- self.assertEqual(f.getparams(), fout.getparams())
- self.assertEqual(f.readframes(5), fout.readframes(5))
-
- def test_compress(self):
- f = self.f = aifc.open(self.sndfilepath)
- fout = self.fout = aifc.open(TESTFN, 'wb')
- fout.aifc()
- fout.setnchannels(f.getnchannels())
- fout.setsampwidth(f.getsampwidth())
- fout.setframerate(f.getframerate())
- fout.setcomptype(b'ULAW', b'foo')
- for frame in range(f.getnframes()):
- fout.writeframes(f.readframes(1))
- fout.close()
- self.assertLess(
- os.stat(TESTFN).st_size,
- os.stat(self.sndfilepath).st_size*0.75,
- )
- fout = self.fout = aifc.open(TESTFN, 'rb')
- f.rewind()
- self.assertEqual(f.getparams()[0:3], fout.getparams()[0:3])
- self.assertEqual(fout.getcomptype(), b'ULAW')
- self.assertEqual(fout.getcompname(), b'foo')
-
- def test_close(self):
- class Wrapfile(object):
- def __init__(self, file):
- self.file = open(file, 'rb')
- self.closed = False
- def close(self):
- self.file.close()
- self.closed = True
- def __getattr__(self, attr): return getattr(self.file, attr)
- testfile = Wrapfile(self.sndfilepath)
- f = self.f = aifc.open(testfile)
- self.assertEqual(testfile.closed, False)
- f.close()
- self.assertEqual(testfile.closed, True)
- testfile = open(TESTFN, 'wb')
- fout = aifc.open(testfile, 'wb')
- self.assertFalse(testfile.closed)
- with self.assertRaises(aifc.Error):
- fout.close()
- self.assertTrue(testfile.closed)
- fout.close() # do nothing
-
+ def test_params_added(self):
+ f = self.f = aifc.open(TESTFN, 'wb')
+ f.aiff()
+ f.setparams((1, 1, 1, 1, b'NONE', b''))
+ f.close()
+
+ f = self.f = aifc.open(TESTFN, 'rb')
+ params = f.getparams()
+ self.assertEqual(params.nchannels, f.getnchannels())
+ self.assertEqual(params.sampwidth, f.getsampwidth())
+ self.assertEqual(params.framerate, f.getframerate())
+ self.assertEqual(params.nframes, f.getnframes())
+ self.assertEqual(params.comptype, f.getcomptype())
+ self.assertEqual(params.compname, f.getcompname())
+
def test_write_header_comptype_sampwidth(self):
for comptype in (b'ULAW', b'ulaw', b'ALAW', b'alaw', b'G722'):
fout = aifc.open(io.BytesIO(), 'wb')