def test_wrong_schema(self):
oid = self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
from psycopg2.extras import CompositeCaster
c = CompositeCaster('type_ii', oid, [('a', 23), ('b', 23), ('c', 23)])
curs = self.conn.cursor()
psycopg2.extensions.register_type(c.typecaster, curs)
curs.execute("select (1,2)::type_ii")
self.assertRaises(psycopg2.DataError, curs.fetchone)
python类DataError()的实例源码
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", (float("inf"),))
self.failUnless(str(s) == "inf", "wrong float quoting: " + str(s))
self.failUnless(type(s) == float, "wrong float conversion: " + repr(s))
s = self.execute("SELECT %s AS foo", (float("-inf"),))
self.failUnless(str(s) == "-inf", "wrong float quoting: " + str(s))
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', '{', '{}}', '{' * 20 + '}' * 20]
for s in ss:
self.assertRaises(psycopg2.DataError,
psycopg2.extensions.STRINGARRAY, b(s), curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.DATE, '2007', self.curs)
self.assertRaises(psycopg2.DataError, self.DATE, '2007-01', self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.TIME, '13', self.curs)
self.assertRaises(psycopg2.DataError, self.TIME, '13:30', self.curs)
def test_parse_incomplete_datetime(self):
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01-01 13', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01-01 13:30', self.curs)
def test_with_error_db(self):
def f():
with self.conn as conn:
curs = conn.cursor()
curs.execute("insert into test_with values ('a')")
self.assertRaises(psycopg2.DataError, f)
self.assertEqual(self.conn.status, ext.STATUS_READY)
self.assert_(not self.conn.closed)
curs = self.conn.cursor()
curs.execute("select * from test_with")
self.assertEqual(curs.fetchall(), [])
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError, e:
self.assertEqual(e.pgcode, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
from psycopg2.extras import CompositeCaster
c = CompositeCaster('type_ii', oid, [('a', 23), ('b', 23), ('c', 23)])
curs = self.conn.cursor()
psycopg2.extensions.register_type(c.typecaster, curs)
curs.execute("select (1,2)::type_ii")
self.assertRaises(psycopg2.DataError, curs.fetchone)
def parse(self, s, curs):
if s is None:
return None
tokens = self.tokenize(s)
if len(tokens) != len(self.atttypes):
raise psycopg2.DataError(
"expecting %d components for the type %s, %d found instead" %
(len(self.atttypes), self.name, len(tokens)))
values = [ curs.cast(oid, token)
for oid, token in zip(self.atttypes, tokens) ]
return self.make(values)
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', '{', '{}}', '{' * 20 + '}' * 20]
for s in ss:
self.assertRaises(psycopg2.DataError,
psycopg2.extensions.STRINGARRAY, b(s), curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.DATE, '2007', self.curs)
self.assertRaises(psycopg2.DataError, self.DATE, '2007-01', self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.TIME, '13', self.curs)
self.assertRaises(psycopg2.DataError, self.TIME, '13:30', self.curs)
def test_parse_incomplete_datetime(self):
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01-01 13', self.curs)
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01-01 13:30', self.curs)
def test_copy_rowcount_error(self):
curs = self.conn.cursor()
curs.execute("insert into tcopy (data) values ('fff')")
self.assertEqual(curs.rowcount, 1)
self.assertRaises(psycopg2.DataError,
curs.copy_from, StringIO('aaa\nbbb\nccc\n'), 'tcopy')
self.assertEqual(curs.rowcount, -1)
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError, e:
self.assertEqual(e.pgcode, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
from psycopg2.extras import CompositeCaster
c = CompositeCaster('type_ii', oid, [('a', 23), ('b', 23), ('c', 23)])
curs = self.conn.cursor()
psycopg2.extensions.register_type(c.typecaster, curs)
curs.execute("select (1,2)::type_ii")
self.assertRaises(psycopg2.DataError, curs.fetchone)
def parse(self, s, curs):
if s is None:
return None
tokens = self.tokenize(s)
if len(tokens) != len(self.atttypes):
raise psycopg2.DataError(
"expecting %d components for the type %s, %d found instead" %
(len(self.atttypes), self.name, len(tokens)))
values = [ curs.cast(oid, token)
for oid, token in zip(self.atttypes, tokens) ]
return self.make(values)
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", (float("inf"),))
self.failUnless(str(s) == "inf", "wrong float quoting: " + str(s))
self.failUnless(type(s) == float, "wrong float conversion: " + repr(s))
s = self.execute("SELECT %s AS foo", (float("-inf"),))
self.failUnless(str(s) == "-inf", "wrong float quoting: " + str(s))
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.DATE, '2007', self.curs)
self.assertRaises(psycopg2.DataError, self.DATE, '2007-01', self.curs)
