class TestPostgresql(unittest.TestCase): def test_gist(self): b = python.custom_db.backends.postgresql.Postgresql( databases=dict( test=dict( minconn=1, maxconn=1 ) ) )
b.connect()
try: c = b.get_connection(use='test')
c2 = c[0]
with c2.cursor() as cur: cur.execute(r''' drop table if exists test; create table test(image_id integer primary key, latent_code custom_cube); create index lc_idx on test using gist(latent_code); ''') c2.commit()
with self.assertRaises(psycopg2.errors.InternalError_): for k in range(10): logging.info('test_postgresql.test_gist, k = %d' % k) cur.execute( r''' insert into test (image_id, latent_code) values (%s, custom_cube(%s)) ''', [ k, [float(x) for x in numpy.random.uniform(0, 1, 512)], ] ) c2.commit() finally: b.put_connection(c2, 'test')