def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1 / 0
def readline(self):
return 1 / 0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception as e:
self.assertTrue('ZeroDivisionError' in str(e))
python类TextIOBase()的实例源码
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1 / 0
def readline(self):
return 1 / 0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def render_samba_configuration(f: TextIOBase, volume_name: str) -> int:
"""
Write the samba configuration file out to disk
:param f: TextIOBase handle to the sambe config file
:param volume_name: str
:return: int of bytes written
"""
bytes_written = 0
bytes_written += f.write("[{}]\n".format(volume_name))
bytes_written += f.write(b"path = /mnt/glusterfs\n"
b"read only = no\n"
b"guest ok = yes\n"
b"kernel share modes = no\n"
b"kernel oplocks = no\n"
b"map archive = no\n"
b"map hidden = no\n"
b"map read only = no\n"
b"map system = no\n"
b"store dos attributes = yes\n")
return bytes_written
def parse(f: IO[Any]) -> Result:
"""
Parse a shellscript and return a ShellScript
:param f: TextIOBase handle to the shellscript file
:return: Result with Ok or Err
"""
comments = []
commands = []
interpreter = ""
buf = f.readlines()
for line in buf:
trimmed = line.strip()
if trimmed.startswith("#!"):
interpreter = trimmed
elif trimmed.startswith("#"):
comments.append(str(trimmed))
else:
# Skip blank lines
if trimmed:
commands.append(str(trimmed))
return Ok(ShellScript(interpreter=interpreter,
comments=comments,
commands=commands))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1 / 0
def readline(self):
return 1 / 0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception as e:
self.assertTrue('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1 / 0
def readline(self):
return 1 / 0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def wrap_file_object(fileobj):
"""Handle differences in Python 2 and 3 around writing bytes."""
# If it's not an instance of IOBase, we're probably using Python 2 and
# that is less finnicky about writing text versus bytes to a file.
if not isinstance(fileobj, io.IOBase):
return fileobj
# At this point we're using Python 3 and that will mangle text written to
# a file written in bytes mode. So, let's check if the file can handle
# text as opposed to bytes.
if isinstance(fileobj, io.TextIOBase):
return fileobj
# Finally, we've determined that the fileobj passed in cannot handle text,
# so we use TextIOWrapper to handle the conversion for us.
return io.TextIOWrapper(fileobj)
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
def skipBOM(fd):
"""
Removes the BOM from UTF-8 files so that we can live in peace.
:param fd: The file descriptor that may or may not have a BOM at the start.
:return: The position after the BOM as reported by fd.tell().
"""
try:
pos = fd.tell()
except IOError:
return 0
if isinstance(fd, io.TextIOBase):
fst = fd.read(len(codecs.BOM_UTF8.decode('utf-8')))
if fst.encode('utf-8') != codecs.BOM_UTF8:
fd.seek(pos)
else:
fst = fd.read(len(codecs.BOM_UTF8))
if fst != codecs.BOM_UTF8:
fd.seek(pos)
return fd.tell()
def deserialize(self, stream, **kwargs):
"""
Deserialize from `PROV-XML <http://www.w3.org/TR/prov-xml/>`_
representation to a :class:`~prov.model.ProvDocument` instance.
:param stream: Input data.
"""
if isinstance(stream, io.TextIOBase):
with io.BytesIO() as buf:
buf.write(stream.read().encode('utf-8'))
buf.seek(0, 0)
xml_doc = etree.parse(buf).getroot()
else:
xml_doc = etree.parse(stream).getroot()
# Remove all comments.
for c in xml_doc.xpath("//comment()"):
p = c.getparent()
p.remove(c)
document = prov.model.ProvDocument()
self.deserialize_subtree(xml_doc, document)
return document
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1/0
def readline(self):
return 1/0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception as e:
self.assertTrue('ZeroDivisionError' in str(e))
def _observe_mdns(reader, output: io.TextIOBase, verbose: bool):
"""Process the given `reader` for `avahi-browse` events.
IO is mostly isolated in this function; the transformation functions
`_observe_all_in_full` and `_observe_resolver_found` can be tested without
having to deal with IO.
:param reader: A context-manager yielding a `io.TextIOBase`.
"""
if verbose:
observer = _observe_all_in_full
else:
observer = _observe_resolver_found
with reader as infile:
events = _extract_mdns_events(infile)
for event in observer(events):
print(json.dumps(event), file=output, flush=True)
def test_copy_from_propagate_error(self):
class BrokenRead(_base):
def read(self, size):
return 1 / 0
def readline(self):
return 1 / 0
curs = self.conn.cursor()
# It seems we cannot do this, but now at least we propagate the error
# self.assertRaises(ZeroDivisionError,
# curs.copy_from, BrokenRead(), "tcopy")
try:
curs.copy_from(BrokenRead(), "tcopy")
except Exception, e:
self.assert_('ZeroDivisionError' in str(e))
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 write(self, f: TextIOBase) -> Result:
# Write the run control class back out to a file
bytes_written = 0
bytes_written += f.write("{}\n".format(self.interpreter))
bytes_written += f.write("\n".join(self.comments))
bytes_written += f.write("\n")
bytes_written += f.write("\n".join(self.commands))
bytes_written += f.write("\n")
return Ok(bytes_written)
def _get_file_object(csvfile, encoding=None):
if isinstance(csvfile, str):
assert encoding, 'encoding required for file path'
return open(csvfile, 'rt', encoding=encoding, newline='') # <- EXIT!
if hasattr(csvfile, 'mode'):
assert 'b' not in csvfile.mode, "File must be open in text mode ('rt')."
elif issubclass(csvfile.__class__, io.IOBase):
assert issubclass(csvfile.__class__, io.TextIOBase), ("Stream object must inherit "
"from io.TextIOBase.")
return csvfile
def _py2_get_file_object(csvfile, encoding):
if isinstance(csvfile, str):
return open(csvfile, 'rb') # <- EXIT!
if hasattr(csvfile, 'mode'):
assert 'b' in csvfile.mode, ("When using Python 2, file must "
"be open in binary mode ('rb').")
elif issubclass(csvfile.__class__, io.IOBase):
assert not issubclass(csvfile.__class__, io.TextIOBase), ("When using Python 2, "
"must use byte stream "
"(not text stream).")
return csvfile
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")
