# This file is Public Domain and may be used without restrictions, # because nobody should have to waste their lives typing this again. import jpype from jpype.types import * from jpype import java import jpype.dbapi2 as dbapi2 import common import datetime import decimal import threading java = jpype.java try: import zlib except ImportError: zlib = None # type: ignore[assignment] db_name = "jdbc:h2:mem:testdb" class ConnectTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def testConnect(self): cx = dbapi2.connect(db_name) self.assertIsInstance(cx, dbapi2.Connection) def testClose(self): cx = dbapi2.connect(db_name) cx.close() # Closing twice is an error with self.assertRaises(dbapi2.ProgrammingError): cx.close() def testScope(self): with dbapi2.connect(db_name) as cx: pass with self.assertRaises(dbapi2.ProgrammingError): cx.close() def testAdapters(self): cx = dbapi2.connect(db_name) self.assertEqual(cx.adapters, dbapi2._default_adapters) cx = dbapi2.connect(db_name, adapters=None) self.assertEqual(cx.adapters, {}) m = {} cx = dbapi2.connect(db_name, adapters=m) self.assertEqual(cx.adapters, m) def testConverters(self): cx = dbapi2.connect(db_name) self.assertEqual(cx.converters, dbapi2._default_converters) cx = dbapi2.connect(db_name, converters=None) self.assertEqual(cx.converters, {}) m = {} cx = dbapi2.connect(db_name, converters=m) self.assertEqual(cx.converters, m) class ConnectionTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def test_commit(self): with dbapi2.connect(db_name) as cx: # Commit must work without a command cx.commit() def test_rollback(self): with dbapi2.connect(db_name) as cx: # Commit must work without a command cx.rollback() def test_cursor(self): with dbapi2.connect(db_name) as cx: self.assertIsInstance(cx.cursor(), dbapi2.Cursor) def testAdapters(self): with dbapi2.connect(db_name) as cx: m = {'foo': 1} cx.adapters = m self.assertEqual(cx._adapters, m) with self.assertRaises(TypeError): cx.adapters = object() def testConverters(self): with dbapi2.connect(db_name) as cx: m = {'foo': 1} cx.converters = m self.assertEqual(cx._converters, m) with self.assertRaises(TypeError): cx.converters = object() def testClosedClass(self): with dbapi2.connect(db_name) as cx: with self.assertRaises(AttributeError): cx.seters = object() def testAutocommitFail(self): with dbapi2.connect(db_name) as cx: cx.autocommit = True with self.assertRaises(dbapi2.NotSupportedError): cx.commit() with self.assertRaises(dbapi2.NotSupportedError): cx.rollback() def test_typeinfo(self): with dbapi2.connect(db_name) as cx: ti = cx.typeinfo self.assertIsInstance(ti, dict) for p, v in ti.items(): self.assertIsInstance(p, str) self.assertIsInstance(v, dbapi2.JDBCType) with self.assertRaises(dbapi2.ProgrammingError): ti = cx.typeinfo def test_connection(self): with dbapi2.connect(db_name) as cx: c = cx.connection self.assertIsInstance(c, java.sql.Connection) def test_getters(self): with dbapi2.connect(db_name) as cx: cx.getters = dbapi2.GETTERS_BY_NAME self.assertEqual(cx._getters, dbapi2.GETTERS_BY_NAME) self.assertEqual(cx.getters, dbapi2.GETTERS_BY_NAME) cx.getters = dbapi2.GETTERS_BY_TYPE self.assertEqual(cx._getters, dbapi2.GETTERS_BY_TYPE) self.assertEqual(cx.getters, dbapi2.GETTERS_BY_TYPE) def test_setters(self): with dbapi2.connect(db_name) as cx: cx.setters = dbapi2.SETTERS_BY_META self.assertEqual(cx._setters, dbapi2.SETTERS_BY_META) self.assertEqual(cx.setters, dbapi2.SETTERS_BY_META) cx.setters = dbapi2.SETTERS_BY_TYPE self.assertEqual(cx._setters, dbapi2.SETTERS_BY_TYPE) self.assertEqual(cx.setters, dbapi2.SETTERS_BY_TYPE) def test_adapters(self): with dbapi2.connect(db_name) as cx: self.assertEqual(cx.adapters, dbapi2._default_adapters) cx.adapters = None self.assertEqual(cx.adapters, {}) m = {} cx.adapters = m self.assertEqual(cx.adapters, m) with self.assertRaises(dbapi2.InterfaceError): cx.adapters = object() def test_converters(self): with dbapi2.connect(db_name) as cx: self.assertEqual(cx.converters, dbapi2._default_converters) cx.converters = None self.assertEqual(cx.converters, {}) m = {} cx.converters = m self.assertEqual(cx.converters, m) with self.assertRaises(dbapi2.InterfaceError): cx.converters = object() class CursorTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table booze") except dbapi2.Error: pass try: cur.execute("drop table barflys") except dbapi2.Error: pass def testCursorIsolation(self): with dbapi2.connect(db_name) as cx: # Make sure cursors created from the same connection have # the documented transaction isolation level cur1 = cx.cursor() cur2 = cx.cursor() cur1.execute("create table booze (name varchar(20))") cur1.execute("insert into booze values ('Victoria Bitter')") cur2.execute("select name from booze") booze = cur2.fetchall() self.assertEqual(len(booze), 1) self.assertEqual(len(booze[0]), 1) self.assertEqual(booze[0][0], "Victoria Bitter") def testDescription(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.execute("create table booze (name varchar(20))") self.assertEqual( cur.description, None, "cursor.description should be none after executing a " "statement that can return no rows (such as create table)", ) cur.execute("select name from booze") self.assertEqual( len(cur.description), 1, "cursor.description describes too many columns" ) self.assertEqual( len(cur.description[0]), 7, "cursor.description[x] tuples must have 7 elements", ) self.assertEqual( cur.description[0][0].lower(), "name", "cursor.description[x][0] must return column name", ) self.assertEqual( cur.description[0][1], dbapi2.STRING, "cursor.description[x][1] must return column type. Got %r" % cur.description[0][1], ) # Make sure self.description gets reset cur.execute("create table barflys (name varchar(20))") self.assertEqual( cur.description, None, "cursor.description not being set to None when executing " "no-result statements (eg. create table)", ) def testRowcount(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.execute("create table booze (name varchar(20))") # This was dropped from the specification # self.assertEqual( # cur.rowcount, # -1, # "cursor.rowcount should be -1 after executing no-result " "statements", # ) cur.execute("insert into booze values ('Victoria Bitter')") self.assertEqual( cur.rowcount, 1, "cursor.rowcount should == number or rows inserted, or " "set to -1 after executing an insert statement", ) cur.execute("select name from booze") self.assertEqual( cur.rowcount, -1, "cursor.rowcount should == number of rows returned, or " "set to -1 after executing a select statement", ) cur.execute("create table barflys (name varchar(20))") # THis test does not match jdbc parameters # self.assertEqual( # cur.rowcount, # -1, # "cursor.rowcount not being reset to -1 after executing " # "no-result statements", # ) def testClose(self): cx = dbapi2.connect(db_name) try: cur = cx.cursor() finally: cx.close() # cursor.execute should raise an Error if called after connection # closed with self.assertRaises(dbapi2.Error): cur.execute("create table booze (name varchar(20))") # connection.commit should raise an Error if called after connection' # closed.' self.assertRaises(dbapi2.Error, cx.commit) # connection.close should raise an Error if called more than once self.assertRaises(dbapi2.Error, cx.close) def test_None(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.execute("create table booze (name varchar(20))") cur.execute("insert into booze values (NULL)") cur.execute("select name from booze") r = cur.fetchall() self.assertEqual(len(r), 1) self.assertEqual(len(r[0]), 1) self.assertEqual(r[0][0], None, "NULL value not returned as None") def testArraysize(self): # Not much here - rest of the tests for this are in test_fetchmany with dbapi2.connect(db_name) as cx, cx.cursor() as cur: self.assertTrue( hasattr(cur, "arraysize"), "cursor.arraysize must be defined" ) def _paraminsert(self, cur): cur.execute("create table booze (name varchar(20))") cur.execute("insert into booze values ('Victoria Bitter')") self.assertTrue(cur.rowcount in (-1, 1)) cur.execute("insert into booze values (?)", ("Cooper's",)) self.assertTrue(cur.rowcount in (-1, 1)) cur.execute("select name from booze") res = cur.fetchall() self.assertEqual(len(res), 2, "cursor.fetchall returned too few rows") beers = [res[0][0], res[1][0]] beers.sort() self.assertEqual( beers[0], "Cooper's", "cursor.fetchall retrieved incorrect data, or data inserted " "incorrectly", ) self.assertEqual( beers[1], "Victoria Bitter", "cursor.fetchall retrieved incorrect data, or data inserted " "incorrectly", ) def test_execute(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: self._paraminsert(cur) def test_executeBad(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: with self.assertRaises(dbapi2.InterfaceError): cur.execute("insert into booze values (?)", object()) with self.assertRaises(dbapi2.InterfaceError): cur.execute(object()) with self.assertRaises(dbapi2.ProgrammingError): cur.execute("inert into booze values (?)", [["?"]]) def test_setinputsizes(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.setinputsizes((25,)) self._paraminsert(cur) # Make sure cursor still works def test_setoutputsize_basic(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.setoutputsize(1000) cur.setoutputsize(2000, 0) self._paraminsert(cur) # Make sure the cursor still works def test_executemany(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.execute("create table booze (name varchar(20))") largs = [("Cooper's",), ("Boag's",)] margs = [{"beer": "Cooper's"}, {"beer": "Boag's"}] cur.executemany("insert into booze values (?)", largs) self.assertEqual( cur.rowcount, 2, "insert using cursor.executemany set cursor.rowcount to " "incorrect value %r" % cur.rowcount, ) cur.execute("select name from booze") res = cur.fetchall() self.assertEqual( len(res), 2, "cursor.fetchall retrieved incorrect number of rows" ) beers = [res[0][0], res[1][0]] beers.sort() self.assertEqual(beers[0], "Boag's", "incorrect data retrieved") self.assertEqual(beers[1], "Cooper's", "incorrect data retrieved") def test_executemanyGenerator(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: def mygen(ls): for i in ls: yield [i] cu.execute("create table booze (name varchar(20))") cu.executemany("insert into booze values (?)", mygen(self.samples)) cu.execute("select * from booze") f = cu.fetchall() for i, v in enumerate(self.samples): self.assertEqual(f[i][0], v) def test_executemanyIterator(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: class myiter: def __init__(self, ls): self.ls = ls self.i = 0 def __iter__(self): return self def __next__(self): if self.i == len(self.ls): raise StopIteration rc = [self.ls[self.i]] self.i += 1 return rc cu.execute("create table booze (name varchar(20))") cu.executemany("insert into booze values (?)", myiter(self.samples)) cu.execute("select * from booze") f = cu.fetchall() for i, v in enumerate(self.samples): self.assertEqual(f[i][0], v) def test_executemanyBad(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze (name varchar(20))") with self.assertRaises(TypeError): cu.executemany("insert into booze values (?)", object()) with self.assertRaises(dbapi2.InterfaceError): cu.executemany("insert into booze values (?)", object()) with self.assertRaises(dbapi2.InterfaceError): cu.executemany(object(), []) cu.executemany("insert into booze values (?)", None) with self.assertRaises(dbapi2.ProgrammingError): cu.executemany("inert into booze values (?)", [['?']]) def test_fetchone(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: # cursor.fetchone should raise an Error if called before # executing a select-type query self.assertRaises(dbapi2.ProgrammingError, cur.fetchone) # cursor.fetchone should raise an Error if called after # executing a query that cannot return rows cur.execute("create table booze (name varchar(20))") self.assertRaises(dbapi2.Error, cur.fetchone) cur.execute("select name from booze") self.assertEqual( cur.fetchone(), None, "cursor.fetchone should return None if a query retrieves " "no rows", ) self.assertEqual(cur.rowcount, -1) # cursor.fetchone should raise an Error if called after # executing a query that cannot return rows cur.execute("insert into booze values ('Victoria Bitter')") self.assertRaises(dbapi2.Error, cur.fetchone) cur.execute("select name from booze") r = cur.fetchone() self.assertEqual( len(r), 1, "cursor.fetchone should have retrieved a single row" ) self.assertEqual( r[0], "Victoria Bitter", "cursor.fetchone retrieved incorrect data" ) self.assertEqual( cur.fetchone(), None, "cursor.fetchone should return None if no more rows available", ) self.assertEqual(cur.rowcount, -1) samples = [ "Carlton Cold", "Carlton Draft", "Mountain Goat", "Redback", "Victoria Bitter", "XXXX", ] def _populate(self): """ Return a list of sql commands to setup the DB for the fetch tests. """ populate = [ "insert into booze values ('{}')".format(s) for s in self.samples ] return populate def test_fetchmany(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: # cursor.fetchmany should raise an Error if called without # issuing a query self.assertRaises(dbapi2.ProgrammingError, cur.fetchmany, 4) cur.execute("create table booze (name varchar(20))") for sql in self._populate(): cur.execute(sql) cur.execute("select name from booze") r = cur.fetchmany() self.assertEqual( len(r), 1, "cursor.fetchmany retrieved incorrect number of rows, " "default of arraysize is one.", ) cur.arraysize = 10 r = cur.fetchmany(3) # Should get 3 rows self.assertEqual( len(r), 3, "cursor.fetchmany retrieved incorrect number of rows" ) r = cur.fetchmany(4) # Should get 2 more self.assertEqual( len(r), 2, "cursor.fetchmany retrieved incorrect number of rows" ) r = cur.fetchmany(4) # Should be an empty sequence self.assertEqual( len(r), 0, "cursor.fetchmany should return an empty sequence after " "results are exhausted", ) self.assertEqual(cur.rowcount, -1) # Same as above, using cursor.arraysize cur.arraysize = 4 cur.execute("select name from booze") r = cur.fetchmany() # Should get 4 rows self.assertEqual( len(r), 4, "cursor.arraysize not being honoured by fetchmany" ) r = cur.fetchmany() # Should get 2 more self.assertEqual(len(r), 2) r = cur.fetchmany() # Should be an empty sequence self.assertEqual(len(r), 0) self.assertEqual(cur.rowcount, -1) cur.arraysize = 6 cur.execute("select name from booze") rows = cur.fetchmany() # Should get all rows self.assertTrue(cur.rowcount in (-1, 6)) self.assertEqual(len(rows), 6) self.assertEqual(len(rows), 6) rows = [r[0] for r in rows] rows.sort() # Make sure we get the right data back out for i in range(0, 6): self.assertEqual( rows[i], self.samples[i], "incorrect data retrieved by cursor.fetchmany", ) rows = cur.fetchmany() # Should return an empty list self.assertEqual( len(rows), 0, "cursor.fetchmany should return an empty sequence if " "called after the whole result set has been fetched", ) self.assertEqual(cur.rowcount, -1) cur.execute("create table barflys (name varchar(20))") cur.execute("select name from barflys") r = cur.fetchmany() # Should get empty sequence self.assertEqual( len(r), 0, "cursor.fetchmany should return an empty sequence if " "query retrieved no rows", ) def test_fetchall(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: # cursor.fetchall should raise an Error if called # without executing a query that may return rows (such # as a select) self.assertRaises(dbapi2.Error, cur.fetchall) cur.execute("create table booze (name varchar(20))") for sql in self._populate(): cur.execute(sql) # cursor.fetchall should raise an Error if called # after executing a statement that cannot return rows self.assertRaises(dbapi2.ProgrammingError, cur.fetchall) cur.execute("select name from booze") rows = cur.fetchall() self.assertEqual(cur.rowcount, -1) self.assertEqual( len(rows), len(self.samples), "cursor.fetchall did not retrieve all rows", ) rows = [r[0] for r in rows] rows.sort() for i in range(0, len(self.samples)): self.assertEqual( rows[i], self.samples[i], "cursor.fetchall retrieved incorrect rows" ) rows = cur.fetchall() self.assertEqual( len(rows), 0, "cursor.fetchall should return an empty list if called " "after the whole result set has been fetched", ) self.assertEqual(cur.rowcount, -1) cur.execute("create table barflys (name varchar(20))") cur.execute("select name from barflys") rows = cur.fetchall() self.assertEqual(cur.rowcount, -1) self.assertEqual( len(rows), 0, "cursor.fetchall should return an empty list if " "a select query returns no rows", ) def test_mixedfetch(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: cur.execute("create table booze (name varchar(20))") for sql in self._populate(): cur.execute(sql) cur.execute("select name from booze") self.assertEqual(cur.rowcount, -1) rows1 = cur.fetchone() rows23 = cur.fetchmany(2) rows4 = cur.fetchone() rows56 = cur.fetchall() self.assertEqual(cur.rowcount, -1) self.assertEqual( len(rows23), 2, "fetchmany returned incorrect number of rows" ) self.assertEqual( len(rows56), 2, "fetchall returned incorrect number of rows" ) rows = [rows1[0]] rows.extend([rows23[0][0], rows23[1][0]]) rows.append(rows4[0]) rows.extend([rows56[0][0], rows56[1][0]]) rows.sort() for i in range(0, len(self.samples)): self.assertEqual( rows[i], self.samples[i], "incorrect data retrieved or inserted" ) def test_nextset(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze (name varchar(20))") booze = ["Wiskey", ] cu.execute("insert into booze(name) values (?)", booze) self.assertEqual(cu.rowcount, 1) cu.execute("select * from booze; select * from booze") self.assertEqual(cu.fetchone(), booze) nxt = cu.nextset() self.assertTrue(nxt in (True, None)) # H2 does not support multiple result sets if nxt: self.assertEqual(cu.fetchone(), booze) def test_lastrowid(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(id identity auto_increment not null, name varchar(255), PRIMARY KEY (id))") cu.execute("insert into booze(name) values('hello')", keys=True) id0 = cu.lastrowid self.assertIsInstance(id0, int) # Call more than once id1 = cu.lastrowid self.assertIsInstance(id1, int) cx.commit() cu.execute("insert into booze(name) values('there')", keys=True) id2 = cu.lastrowid self.assertIsInstance(id2, int) cu.execute("select * from booze") f = cu.fetchall() self.assertEqual(cu.lastrowid, None) self.assertEqual(f[0][0], id0) self.assertEqual(f[1][0], id2) def test_lastrowidMany(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(id identity auto_increment not null, name varchar(255), PRIMARY KEY (id))") cu.executemany("insert into booze(name) values(?)", [['Redback'], ['Fat Yak']], keys=True) ids = cu.lastrowid self.assertIsInstance(ids, list) self.assertEqual(len(ids), 2) cu.execute("select * from booze") f = cu.fetchall() self.assertEqual(f[0][0], ids[0]) self.assertEqual(f[1][0], ids[1]) def test_iter(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: ls = [['Tooheys Old'], ['Nine Tales Original Amber Ale'], ['Dogbolter']] cu.execute("create table booze(name varchar(255))") cu.executemany("insert into booze(name) values(?)", ls, keys=True) # make sure we throw if there is no result set with self.assertRaises(dbapi2.ProgrammingError): for booze in cu: pass cu.execute("select * from booze") for i, booze in enumerate(cu): self.assertEqual(ls[i], booze) def testNoAdapter(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(name varchar(50))") with self.assertRaises(dbapi2.InterfaceError): cu.execute("insert into booze(name) values(?)", [object()]) def testBadParameters(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(name varchar(50))") with self.assertRaises(dbapi2.InterfaceError): cu.execute("insert into booze(name) values(?)", object()) def test_callproc(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: r = cu.callproc("lower", ("FOO",)) self.assertEqual(len(r), 1) self.assertEqual(r[0], "FOO") r = cu.fetchall() self.assertEqual(len(r), 1) self.assertEqual(len(r[0]), 1) self.assertEqual(r[0][0], "foo") def test_callprocBad(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: with self.assertRaises(dbapi2.ProgrammingError): r = cu.callproc("not_there", ("FOO",)) with self.assertRaises(dbapi2.InterfaceError): r = cu.callproc(object(), ("FOO",)) with self.assertRaises(dbapi2.InterfaceError): r = cu.callproc("lower", (object(),)) with self.assertRaises(dbapi2.InterfaceError): r = cu.callproc("lower", object()) def test_close(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.close() with self.assertRaises(dbapi2.ProgrammingError): cu.close() def test_parameters(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: with self.assertRaises(dbapi2.ProgrammingError): p = cu.parameters cu.execute("create table booze(name varchar(50))") cu.execute("insert into booze(name) values(?)", ('Hahn Super Dry',)) p = cu.parameters self.assertEqual(p[0][0:2], ('VARCHAR', dbapi2.VARCHAR)) def test_resultSet(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(name varchar(50))") cu.execute("insert into booze(name) values(?)", ('Hahn Super Dry',)) cu.execute("select * from booze") rs = cu.resultSet self.assertIsInstance(rs, java.sql.ResultSet) def test_fetchClosed(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table booze(name varchar(50))") cu.execute("insert into booze(name) values(?)", ('Hahn Super Dry',)) cu.execute("select * from booze") cu.close() with self.assertRaises(dbapi2.ProgrammingError): cu.fetchone() with self.assertRaises(dbapi2.ProgrammingError): cu.fetchmany() with self.assertRaises(dbapi2.ProgrammingError): cu.fetchall() self.assertEqual(cu.resultSet, None) def test_executeBadParameters(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: def mygen(ls): for i in ls: yield i cu.execute("create table booze(name varchar(50), price integer)") with self.assertRaises(dbapi2.ProgrammingError): cu.executemany("insert into booze(name,price) values(?,?)", [('Hahn Super Dry', 2), ('Coors',)]) with self.assertRaises(dbapi2.InterfaceError): cu.execute("insert into booze(name,price) values(?,?)", {'name': 'Budweiser', 'price': 10}) with self.assertRaises(dbapi2.ProgrammingError): cu.executemany("insert into booze(name,price) values(?,?)", mygen([['Budweiser']])) cu.execute("insert into booze(name,price) values(?,?)", mygen(['Pabst Blue Robot', 20000])) with self.assertRaises(dbapi2.ProgrammingError): cu.execute("insert into booze(name,price) values(?,?)", mygen(['Budweiser'])) with self.assertRaises(dbapi2.InterfaceError): cu.execute("insert into booze(name,price) values(?,?)", 'Budweiser') with self.assertRaises(dbapi2.ProgrammingError): cu.execute("insert into booze(name,price) values(?,?)", mygen(['Budweiser', 2, 'no'])) with self.assertRaises(dbapi2.InterfaceError): cu.execute("insert into booze(name,price) values(?,?)", object()) @common.unittest.skipUnless(zlib, "requires zlib") class AdapterTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") self.testdata = b"abcdefg" * 10 self.params = memoryview(zlib.compress(self.testdata)) def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass def testOnConnector(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(255))") cx.adapters[object] = lambda x: "none" cu.execute("insert into test(name) values(?)", [object()]) cu.execute("select * from test") f = cu.fetchall() self.assertEqual(f[0][0], "none") @common.unittest.skipUnless(zlib, "requires zlib") class ConverterTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") self.testdata = b"abcdefg" * 10 self.params = memoryview(zlib.compress(self.testdata)) def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass @staticmethod def convert(s): return zlib.decompress(s) def testConvertersBad(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") with self.assertRaises(dbapi2.ProgrammingError): cu.fetchone(converters=[]) cu.execute("select name from test") with self.assertRaises(dbapi2.InterfaceError): cu.fetchone(converters=[int]) cu.execute("select name from test") with self.assertRaises(dbapi2.ProgrammingError): cu.fetchone(converters=[str, str]) def testConvertersNone(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") f = cu.fetchone(converters=None) self.assertIsInstance(f[0], java.lang.String) cu.execute("select name from test") def testConvertersPositional(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") f = cu.fetchone(converters=[str]) self.assertIsInstance(f[0], str) cu.execute("select name from test") def test_fetchoneTypesPositional(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") f = cu.fetchone(types=[dbapi2.STRING]) self.assertIsInstance(f[0], str) def test_fetchmanyTypesPositional(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") f = cu.fetchmany(types=[dbapi2.STRING]) self.assertIsInstance(f[0][0], str) def test_fetchallTypesPositional(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") f = cu.fetchall(types=[dbapi2.STRING]) self.assertIsInstance(f[0][0], str) def testTypesPositionalBAD(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name varchar(20))") cu.executemany("insert into test(name) values(?)", [['alice'], ['bob'], ['charlie']]) cu.execute("select name from test") with self.assertRaises(dbapi2.ProgrammingError): f = cu.fetchone(types=[dbapi2.STRING, dbapi2.STRING]) with self.assertRaises(dbapi2.ProgrammingError): f = cu.fetchone(types=[]) @common.unittest.skipUnless(zlib, "requires zlib") class GettersTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") self.testdata = b"abcdefg" * 10 self.params = memoryview(zlib.compress(self.testdata)) def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass class TransactionsTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass def testTransactionRollbackCreate(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cx.commit() # needed before starting transactions cu.execute("begin") cu.execute("create table test(name VARCHAR(10))") cx.rollback() # with self.assertRaises(dbapi2.ProgrammingError): cu.execute("select * from test") def testTransactionRollbackInsert(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cx.commit() # needed before starting transactions cu.execute("create table test(name VARCHAR(10))") cx.commit() cu.execute("begin") cu.execute("insert into test(name) values('alice')") cx.rollback() # Alice should go away result = cu.execute("select * from test").fetchall() self.assertEqual(result, []) def testTransactionRollbackClose(self): with dbapi2.connect(db_name) as cx2: with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cx.commit() # needed before starting transactions cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('alice')") cx.commit() cu.execute("begin") cu.execute("insert into test(name) values('bob')") result = cu.execute("select * from test").fetchall() self.assertEqual(result, [['alice'], ['bob']]) # Bob should go away with cx2.cursor() as cu: result = cu.execute("select * from test").fetchall() self.assertEqual(result, [['alice']]) class TypeTestCase(common.JPypeTestCase): """ This test is db dependent, but needed to check all code paths. """ def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass def testTime(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME TIME)") cu.execute("select * from test") cu.execute("insert into test(NAME) values('12:05:10')") f = cu.execute('select * from test').fetchone() self.assertIsInstance(f[0], datetime.time) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", f) f2 = cu.execute('select * from test').fetchone() self.assertEqual(f2, f) cu.execute("delete from test") # Test with Java t1 = jpype.java.sql.Time(1, 2, 3) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.time(1, 2, 3)) f4 = cu.execute('select * from test').fetchone(converters=None) self.assertEqual(f4[0], t1) self.assertIsInstance(f4[0], jpype.java.sql.Time) cu.execute("delete from test") # Test with dbapi2.Time t1 = dbapi2.Time(6, 21, 32) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.time(6, 21, 32)) def testDate(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME DATE)") cu.execute("select * from test") cu.execute("insert into test(NAME) values('2012-02-05')") f = cu.execute('select * from test').fetchone() self.assertIsInstance(f[0], datetime.date) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", f) f2 = cu.execute('select * from test').fetchone() self.assertEqual(f2, f) cu.execute("delete from test") # Test with Java t1 = java.sql.Date(1972 - 1900, 4 - 1, 1) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.date(1972, 4, 1)) f4 = cu.execute('select * from test').fetchone(converters=None) self.assertEqual(f4[0], t1) self.assertIsInstance(f4[0], java.sql.Date) cu.execute("delete from test") # Test with dbapi2.Date t1 = dbapi2.Date(2020, 5, 21) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.date(2020, 5, 21)) def testTimestamp(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME TIMESTAMP)") cu.execute("select * from test") cu.execute("insert into test(NAME) values('2012-02-05 12:02:45.123')") f = cu.execute('select * from test').fetchone() self.assertIsInstance(f[0], datetime.date) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", f) f2 = cu.execute('select * from test').fetchone() self.assertEqual(f2, f) cu.execute("delete from test") # Test with Java t1 = java.sql.Timestamp(1972 - 1900, 4 - 1, 1, 12, 2, 45, 123456000) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.datetime(1972, 4, 1, 12, 2, 45, 123456)) f4 = cu.execute('select * from test').fetchone(converters=None) self.assertEqual(f4[0], t1) self.assertIsInstance(f4[0], java.sql.Timestamp) cu.execute("delete from test") # Test with dbapi2.Date t1 = dbapi2.Timestamp(2020, 5, 21, 3, 4, 5, 123122000) cu.execute("insert into test(NAME) values(?)", [t1]) f3 = cu.execute('select * from test').fetchone() self.assertEqual(f3[0], datetime.datetime(2020, 5, 21, 3, 4, 5, 123122)) def _testInt(self, tp, desc, jtype, null=True): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME %s)" % tp) cu.execute("insert into test(NAME) values(?)", [123]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], 123) self.assertIsInstance(f[0], int) self.assertEqual(cu.description[0][0:2], desc) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], 123) self.assertIsInstance(f[0], jtype) cu.execute("delete from test") if null: cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) def testTinyInt(self): self._testInt('TINYINT', ('NAME', 'TINYINT'), jpype.JShort) def testBigInt(self): self._testInt('BIGINT', ('NAME', 'BIGINT'), jpype.JLong) def testIdentity(self): self._testInt('IDENTITY', ('NAME', 'BIGINT'), jpype.JLong, False) def testInteger(self): self._testInt('INTEGER', ('NAME', 'INTEGER'), jpype.JInt) def testSmallInt(self): self._testInt('SMALLINT', ('NAME', 'SMALLINT'), jpype.JShort) def _testFloat(self, tp, desc, jtype): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME %s)" % tp) cu.execute("insert into test(NAME) values(?)", [1.25]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], 1.25) self.assertIsInstance(f[0], float) self.assertEqual(cu.description[0][0:2], desc) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], 1.25) self.assertIsInstance(f[0], jtype) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) def testFloat(self): self._testFloat('FLOAT', ('NAME', 'DOUBLE'), jpype.JDouble) def testDouble(self): self._testFloat('DOUBLE', ('NAME', 'DOUBLE'), jpype.JDouble) def testReal(self): self._testFloat('REAL', ('NAME', 'REAL'), jpype.JFloat) def _testNumeric(self, tp, desc, jtype): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME %s(5,2))" % tp) cu.execute("insert into test(NAME) values(?)", [1.25]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], 1.25) self.assertIsInstance(f[0], decimal.Decimal) self.assertEqual(cu.description[0][0:2], desc) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], decimal.Decimal(1.25)) self.assertIsInstance(f[0], jtype) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) def testNumeric(self): self._testNumeric('NUMERIC', ('NAME', 'DECIMAL'), java.math.BigDecimal) def testDecimal(self): self._testNumeric('DECIMAL', ('NAME', 'DECIMAL'), java.math.BigDecimal) def _testBinary(self, tp, desc, jtype): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME %s)" % tp) v = bytes([1, 2, 3]) cu.execute("insert into test(NAME) values(?)", [v]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], v) self.assertIsInstance(f[0], bytes) self.assertEqual(cu.description[0][0:2], desc) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(bytes(f[0]), v) self.assertIsInstance(f[0], jtype) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) def testBinary(self): self._testBinary('BINARY(3)', ('NAME', 'VARBINARY'), JArray(JByte)) def testBlob(self): self._testBinary('BLOB', ('NAME', 'BLOB'), JArray(JByte)) def testLongVarBinary(self): self._testBinary('LONGVARBINARY', ('NAME', 'VARBINARY'), JArray(JByte)) def testVarBinary(self): self._testBinary('VARBINARY(10)', ('NAME', 'VARBINARY'), JArray(JByte)) def _testChars(self, tp, desc, jtype, v="hello"): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME %s)" % tp) cu.execute("insert into test(NAME) values(?)", [v]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], v) self.assertIsInstance(f[0], str) self.assertEqual(cu.description[0][0:2], desc) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], v) self.assertIsInstance(f[0], jtype) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) def testChar(self): self._testChars('CHAR', ('NAME', 'CHAR'), java.lang.String, 'a') def testVarChar(self): self._testChars('VARCHAR', ('NAME', 'VARCHAR'), java.lang.String) def testLongVarChar(self): self._testChars('LONGVARCHAR', ('NAME', 'VARCHAR'), java.lang.String) def testClob(self): self._testChars('CLOB', ('NAME', 'CLOB'), java.lang.String) def testBoolean(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(NAME BOOLEAN)") cu.execute("insert into test(NAME) values(?)", [True]) f = cu.execute("select * from test").fetchone() self.assertEqual(f[0], True) self.assertIsInstance(f[0], int) self.assertEqual(cu.description[0][0:2], ('NAME', 'BOOLEAN')) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], True) self.assertIsInstance(f[0], type(True)) cu.execute("delete from test") cu.execute("insert into test(NAME) values(?)", [None]) f = cu.execute("select * from test").fetchone(converters=None) self.assertEqual(f[0], None) class OtherTestCase(common.JPypeTestCase): """ This test is db dependent, but needed to check all code paths. """ def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass def testJSON(self): got = [] class MyType(dbapi2.JDBCType): def get(self, rs, column, st): got.append(column) return dbapi2.JDBCType.get(self, rs, column, st) # Register a new type JSON = MyType("JSON", None) with dbapi2.connect(db_name, getters=dbapi2.GETTERS_BY_TYPE) as cx, cx.cursor() as cu: cu.execute("create table test(name JSON)") cu.execute("insert into test(name) values('{age: 31}')") f = cu.execute("select * from test").fetchone() self.assertEqual(got, [1]) with dbapi2.connect(db_name, getters=dbapi2.GETTERS_BY_NAME) as cx, cx.cursor() as cu: cu.execute("create table test(name JSON)") cu.execute("insert into test(name) values('{age: 31}')") f = cu.execute("select * from test").fetchone() self.assertEqual(got, [1, 1]) del dbapi2._registry[JSON] class ThreadingTestCase(common.JPypeTestCase): def setUp(self): common.JPypeTestCase.setUp(self) if common.fast: raise common.unittest.SkipTest("fast") def tearDown(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cur: try: cur.execute("drop table test") except dbapi2.Error: pass def _testThread(self, cu, cmd, args, exc): error = [] def run(): try: cmd(*args) except Exception as ex: error.append(ex) t = threading.Thread(target=run) t.start() t.join() self.assertEqual(len(error), 1) with self.assertRaises(exc): raise error[0] def test_execute(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") self._testThread(cu, cu.execute, ("insert into test(name) values ('a')",), dbapi2.ProgrammingError) def test_executemany(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") self._testThread(cu, cu.executemany, ("insert into test(name) values ('?')", [[1]]), dbapi2.ProgrammingError) def test_fetchone(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('a')") self._testThread(cu, cu.fetchone, (), dbapi2.ProgrammingError) def test_fetchmany(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('a')") self._testThread(cu, cu.fetchmany, (), dbapi2.ProgrammingError) def test_fetchall(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('a')") self._testThread(cu, cu.fetchall, (), dbapi2.ProgrammingError) def test_close(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('a')") self._testThread(cu, cu.close, (), dbapi2.ProgrammingError) def test_callproc(self): with dbapi2.connect(db_name) as cx, cx.cursor() as cu: cu.execute("create table test(name VARCHAR(10))") cu.execute("insert into test(name) values('a')") self._testThread(cu, cu.callproc, ("lower", ["FOO"]), dbapi2.ProgrammingError)