def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj):
pass
def getquoted(self):
return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b"SELECT NULL;", s)
finally:
ext.register_adapter(type(None), orig_adapter)
python类adapters()的实例源码
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj):
pass
def getquoted(self):
return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b"SELECT NULL;", s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj):
pass
def getquoted(self):
return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b"SELECT NULL;", s)
finally:
ext.register_adapter(type(None), orig_adapter)
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b("SELECT NULL;"), s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj): pass
def getquoted(self): return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b"SELECT NULL;", s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self):
# the None adapter is not actually invoked in regular adaptation
class WonkyAdapter(object):
def __init__(self, obj):
pass
def getquoted(self):
return "NOPE!"
curs = self.conn.cursor()
orig_adapter = ext.adapters[type(None), ext.ISQLQuote]
try:
ext.register_adapter(type(None), WonkyAdapter)
self.assertEqual(ext.adapt(None).getquoted(), "NOPE!")
s = curs.mogrify("SELECT %s;", (None,))
self.assertEqual(b"SELECT NULL;", s)
finally:
ext.register_adapter(type(None), orig_adapter)
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""")
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assertTrue(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
cur.execute("select %s", ([r, r, r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""")
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r, r, r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertQuotedEqual(
curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""")
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assertTrue(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
cur.execute("select %s", ([r, r, r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
