def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
python类TextIOBase()的实例源码
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def test_misc(self):
shell = MockShell()
f = PseudoOutputFile(shell, 'stdout', 'utf-8')
self.assertIsInstance(f, io.TextIOBase)
self.assertEqual(f.encoding, 'utf-8')
self.assertIsNone(f.errors)
self.assertIsNone(f.newlines)
self.assertEqual(f.name, '<stdout>')
self.assertFalse(f.closed)
self.assertTrue(f.isatty())
self.assertFalse(f.readable())
self.assertTrue(f.writable())
self.assertFalse(f.seekable())
def test_misc(self):
shell = MockShell()
f = PseudoInputFile(shell, 'stdin', 'utf-8')
self.assertIsInstance(f, io.TextIOBase)
self.assertEqual(f.encoding, 'utf-8')
self.assertIsNone(f.errors)
self.assertIsNone(f.newlines)
self.assertEqual(f.name, '<stdin>')
self.assertFalse(f.closed)
self.assertTrue(f.isatty())
self.assertTrue(f.readable())
self.assertFalse(f.writable())
self.assertFalse(f.seekable())
def test_misc(self):
shell = MockShell()
f = PseudoOutputFile(shell, 'stdout', 'utf-8')
self.assertIsInstance(f, io.TextIOBase)
self.assertEqual(f.encoding, 'utf-8')
self.assertIsNone(f.errors)
self.assertIsNone(f.newlines)
self.assertEqual(f.name, '<stdout>')
self.assertFalse(f.closed)
self.assertTrue(f.isatty())
self.assertFalse(f.readable())
self.assertTrue(f.writable())
self.assertFalse(f.seekable())
def test_misc(self):
shell = MockShell()
f = PseudoInputFile(shell, 'stdin', 'utf-8')
self.assertIsInstance(f, io.TextIOBase)
self.assertEqual(f.encoding, 'utf-8')
self.assertIsNone(f.errors)
self.assertIsNone(f.newlines)
self.assertEqual(f.name, '<stdin>')
self.assertFalse(f.closed)
self.assertTrue(f.isatty())
self.assertTrue(f.readable())
self.assertFalse(f.writable())
self.assertFalse(f.seekable())
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1 / 0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def _gettextwriter(out, encoding):
if out is None:
import sys
return sys.stdout
if isinstance(out, io.TextIOBase):
# use a text writer as is
return out
# wrap a binary writer with TextIOWrapper
if isinstance(out, io.RawIOBase):
# Keep the original file open when the TextIOWrapper is
# destroyed
class _wrapper:
__class__ = out.__class__
def __getattr__(self, name):
return getattr(out, name)
buffer = _wrapper()
buffer.close = lambda: None
else:
# This is to handle passed objects that aren't in the
# IOBase hierarchy, but just have a write method
buffer = io.BufferedIOBase()
buffer.writable = lambda: True
buffer.write = out.write
try:
# TextIOWrapper uses this methods to determine
# if BOM (for UTF-16, etc) should be added
buffer.seekable = out.seekable
buffer.tell = out.tell
except AttributeError:
pass
return io.TextIOWrapper(buffer, encoding=encoding,
errors='xmlcharrefreplace',
newline='\n',
write_through=True)
def _workspace(cls) -> TextIOBase:
return NamedTemporaryFile()
def _open(cls, fileobj: BytesIO, mode: str='rb') -> TextIOBase:
return GzipFile(fileobj=fileobj, mode=mode)
def _download_to_stream(self, blobname: str, container: str,
stream: TextIOBase) -> bool:
try:
self._azure_client.get_blob_to_stream(container, blobname, stream)
except AzureMissingResourceHttpError:
return False
else:
return True
def _upload_from_stream(self, blobname: str, stream: TextIOBase):
self._azure_client.create_blob_from_stream(self._container,
blobname, stream)
def _upload_from_stream(self, blobname: str, stream: TextIOBase):
upload_directory = path.join(getenv('AZURE_ROOT'), self._container)
makedirs(upload_directory, exist_ok=True)
local_filename = path.join(upload_directory, blobname)
with open(local_filename, 'wb') as fobj:
copyfileobj(stream, fobj)
def test_copy_to_propagate_error(self):
class BrokenWrite(_base):
def write(self, data):
return 1/0
curs = self.conn.cursor()
curs.execute("insert into tcopy values (10, 'hi')")
self.assertRaises(ZeroDivisionError,
curs.copy_to, BrokenWrite(), "tcopy")
def is_fileobj(f):
""" Check if an object `f` is intance of FileIO object created
by `open()`"""
return isinstance(f, io.TextIOBase) or \
isinstance(f, io.BufferedIOBase) or \
isinstance(f, io.RawIOBase) or \
isinstance(f, io.IOBase)
def test_open_common(self):
p = self.cls(BASE)
with (p / 'fileA').open('r') as f:
self.assertIsInstance(f, io.TextIOBase)
self.assertEqual(f.read(), "this is file A\n")
with (p / 'fileA').open('rb') as f:
self.assertIsInstance(f, io.BufferedIOBase)
self.assertEqual(f.read().strip(), b"this is file A")
with (p / 'fileA').open('rb', buffering=0) as f:
self.assertIsInstance(f, io.RawIOBase)
self.assertEqual(f.read().strip(), b"this is file A")
