346 lines
12 KiB
Python
346 lines
12 KiB
Python
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 UnseekableIO(io.FileIO):
|
|
def tell(self):
|
|
raise io.UnsupportedOperation
|
|
|
|
def seek(self, *args, **kwargs):
|
|
raise io.UnsupportedOperation
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
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_write_bytearray(self):
|
|
f = self.create_file(TESTFN)
|
|
f.setnframes(self.nframes)
|
|
f.writeframes(bytearray(self.frames))
|
|
f.close()
|
|
|
|
self.check_file(TESTFN, self.nframes, self.frames)
|
|
|
|
def test_write_array(self):
|
|
f = self.create_file(TESTFN)
|
|
f.setnframes(self.nframes)
|
|
f.writeframes(array.array('h', self.frames))
|
|
f.close()
|
|
|
|
self.check_file(TESTFN, self.nframes, self.frames)
|
|
|
|
def test_write_memoryview(self):
|
|
f = self.create_file(TESTFN)
|
|
f.setnframes(self.nframes)
|
|
f.writeframes(memoryview(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)
|
|
|
|
def test_unseekable_read(self):
|
|
with self.create_file(TESTFN) as f:
|
|
f.setnframes(self.nframes)
|
|
f.writeframes(self.frames)
|
|
|
|
with UnseekableIO(TESTFN, 'rb') as testfile:
|
|
self.check_file(testfile, self.nframes, self.frames)
|
|
|
|
def test_unseekable_write(self):
|
|
with UnseekableIO(TESTFN, 'wb') as testfile:
|
|
with self.create_file(testfile) as f:
|
|
f.setnframes(self.nframes)
|
|
f.writeframes(self.frames)
|
|
|
|
self.check_file(TESTFN, self.nframes, self.frames)
|
|
|
|
def test_unseekable_incompleted_write(self):
|
|
with UnseekableIO(TESTFN, 'wb') as testfile:
|
|
testfile.write(b'ababagalamaga')
|
|
f = self.create_file(testfile)
|
|
f.setnframes(self.nframes + 1)
|
|
try:
|
|
f.writeframes(self.frames)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
f.close()
|
|
except OSError:
|
|
pass
|
|
|
|
with open(TESTFN, 'rb') as testfile:
|
|
self.assertEqual(testfile.read(13), b'ababagalamaga')
|
|
self.check_file(testfile, self.nframes + 1, self.frames)
|
|
|
|
def test_unseekable_overflowed_write(self):
|
|
with UnseekableIO(TESTFN, 'wb') as testfile:
|
|
testfile.write(b'ababagalamaga')
|
|
f = self.create_file(testfile)
|
|
f.setnframes(self.nframes - 1)
|
|
try:
|
|
f.writeframes(self.frames)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
f.close()
|
|
except OSError:
|
|
pass
|
|
|
|
with open(TESTFN, 'rb') as testfile:
|
|
self.assertEqual(testfile.read(13), b'ababagalamaga')
|
|
framesize = self.nchannels * self.sampwidth
|
|
self.check_file(testfile, self.nframes - 1, self.frames[:-framesize])
|
|
|
|
|
|
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):
|
|
with open(self.sndfilepath, 'rb') as testfile:
|
|
f = self.f = self.module.open(testfile)
|
|
self.assertFalse(testfile.closed)
|
|
f.close()
|
|
self.assertEqual(testfile.closed, self.close_fd)
|
|
with open(TESTFN, 'wb') as testfile:
|
|
fout = self.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')
|
|
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)
|