Briq/jaydebeapi
6
0 Fork 0
Code Issues Pull Requests Releases Wiki Activity

Run database specific tests in separate Travis builds

Browse Source
This commit is contained in:
baztian committed 2014-02-12 23:12:51 +01:00
1 parent 235c0ea7ec
commit eb57a04453
7 files changed
+16 -19
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/test/data/create.sql
-13
View File
@@ -1,13 +0,0 @@
create table Account (
"ACCOUNT_ID" TIMESTAMP default CURRENT_TIMESTAMP not null,
"ACCOUNT_NO" INTEGER not null,
"BALANCE" DECIMAL default 0.0 not null,
"BLOCKING" DECIMAL,
"DBL_COL" DOUBLE,
"OPENED_AT" DATE,
"VALID" BOOLEAN,
"PRODUCT_NAME" VARCHAR(50),
"STUFF" BLOB,
primary key ("ACCOUNT_ID")
);
src/test/data/create_hsqldb.sql
-12
View File
@@ -1,12 +0,0 @@
create table Account (
"ACCOUNT_ID" TIMESTAMP default CURRENT_TIMESTAMP not null,
"ACCOUNT_NO" INTEGER not null,
"BALANCE" DECIMAL default 0.0 not null,
"BLOCKING" DECIMAL,
"DBL_COL" DOUBLE,
"OPENED_AT" DATE,
"VALID" BOOLEAN,
"PRODUCT_NAME" VARCHAR(50),
primary key ("ACCOUNT_ID")
);
src/test/data/insert.sql
-2
View File
@@ -1,2 +0,0 @@
insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING) values ('2009-09-10 14:15:22.123456', 18, 12.4, null);
insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING) values ('2009-09-11 14:15:22.123456', 19, 12.9, 1);
src/test/test_integration.py
-243
View File
@@ -1,243 +0,0 @@
#-*- coding: utf-8 -*-
# Copyright 2010 Bastian Bowe
#
# This file is part of JayDeBeApi.
# JayDeBeApi is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# JayDeBeApi is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with JayDeBeApi. If not, see
# <http://www.gnu.org/licenses/>.
import jaydebeapi
import os
import sys
import threading
import unittest2 as unittest
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
def is_jython():
return sys.platform.lower().startswith('java')
class IntegrationTestBase(object):
def sql_file(self, filename):
f = open(filename, 'r')
try:
lines = f.readlines()
finally:
f.close()
stmt = []
stmts = []
for i in lines:
stmt.append(i)
if ";" in i:
stmts.append(" ".join(stmt))
stmt = []
cursor = self.conn.cursor()
for i in stmts:
cursor.execute(i)
def setUp(self):
(self.dbapi, self.conn) = self.connect()
self.setUpSql()
def setUpSql(self):
raise NotImplementedError
def connect(self):
raise NotImplementedError
def tearDown(self):
cursor = self.conn.cursor()
cursor.execute("drop table ACCOUNT");
self.conn.close()
def test_execute_and_fetch_no_data(self):
cursor = self.conn.cursor()
stmt = "select * from ACCOUNT where ACCOUNT_ID is null"
cursor.execute(stmt)
self.assertEqual(cursor.fetchall(), [])
def test_execute_and_fetch(self):
cursor = self.conn.cursor()
cursor.execute("select ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING " \
"from ACCOUNT")
result = cursor.fetchall()
self.assertEqual(result, [(u'2009-09-10 14:15:22.123456', 18, 12.4, None),
(u'2009-09-11 14:15:22.123456', 19, 12.9, 1)])
def test_execute_and_fetch_parameter(self):
cursor = self.conn.cursor()
cursor.execute("select ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING " \
"from ACCOUNT where ACCOUNT_NO = ?", (18,))
result = cursor.fetchall()
self.assertEqual(result, [(u'2009-09-10 14:15:22.123456', 18, 12.4, None)])
def test_execute_and_fetchone(self):
cursor = self.conn.cursor()
cursor.execute("select ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING " \
"from ACCOUNT order by ACCOUNT_NO")
result = cursor.fetchone()
self.assertEqual(result, (u'2009-09-10 14:15:22.123456', 18, 12.4, None))
cursor.close()
def test_execute_reset_description_without_execute_result(self):
"""Excpect the descriptions property being reset when no query
has been made via execute method.
"""
cursor = self.conn.cursor()
cursor.execute("select * from ACCOUNT")
self.assertIsNotNone(cursor.description)
cursor.fetchone()
cursor.execute("delete from ACCOUNT")
self.assertIsNone(cursor.description)
def test_execute_and_fetchone_after_end(self):
cursor = self.conn.cursor()
cursor.execute("select * from ACCOUNT where ACCOUNT_NO = ?", (18,))
cursor.fetchone()
result = cursor.fetchone()
self.assertIsNone(result)
def test_execute_and_fetchmany(self):
cursor = self.conn.cursor()
cursor.execute("select ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING " \
"from ACCOUNT order by ACCOUNT_NO")
result = cursor.fetchmany()
self.assertEqual(result, [(u'2009-09-10 14:15:22.123456', 18, 12.4, None)])
# TODO: find out why this cursor has to be closed in order to
# let this test work with sqlite if __del__ is not overridden
# in cursor
# cursor.close()
def test_executemany(self):
cursor = self.conn.cursor()
stmt = "insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE) " \
"values (?, ?, ?)"
parms = (
( '2009-09-11 14:15:22.123450', 20, 13.1 ),
( '2009-09-11 14:15:22.123451', 21, 13.2 ),
( '2009-09-11 14:15:22.123452', 22, 13.3 ),
)
cursor.executemany(stmt, parms)
self.assertEqual(cursor.rowcount, 3)
def test_execute_types(self):
cursor = self.conn.cursor()
stmt = "insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE, " \
"BLOCKING, DBL_COL, OPENED_AT, VALID, PRODUCT_NAME) " \
"values (?, ?, ?, ?, ?, ?, ?, ?)"
d = self.dbapi
account_id = d.Timestamp(2010, 01, 26, 14, 31, 59)
account_no = 20
balance = 1.2
blocking = 10.0
dbl_col = 3.5
opened_at = d.Date(2008, 02, 27)
valid = 1
product_name = u'Savings account'
parms = (account_id, account_no, balance, blocking, dbl_col,
opened_at, valid, product_name)
cursor.execute(stmt, parms)
stmt = "select ACCOUNT_ID, ACCOUNT_NO, BALANCE, BLOCKING, " \
"DBL_COL, OPENED_AT, VALID, PRODUCT_NAME " \
"from ACCOUNT where ACCOUNT_NO = ?"
parms = (20, )
cursor.execute(stmt, parms)
result = cursor.fetchone()
cursor.close()
exp = ( '2010-01-26 14:31:59', account_no, balance, blocking,
dbl_col, '2008-02-27', valid, product_name )
self.assertEqual(result, exp)
def test_execute_different_rowcounts(self):
cursor = self.conn.cursor()
stmt = "insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE) " \
"values (?, ?, ?)"
parms = (
( '2009-09-11 14:15:22.123450', 20, 13.1 ),
( '2009-09-11 14:15:22.123452', 22, 13.3 ),
)
cursor.executemany(stmt, parms)
self.assertEqual(cursor.rowcount, 2)
parms = ( '2009-09-11 14:15:22.123451', 21, 13.2 )
cursor.execute(stmt, parms)
self.assertEqual(cursor.rowcount, 1)
cursor.execute("select * from ACCOUNT")
self.assertEqual(cursor.rowcount, -1)
class SqliteTestBase(IntegrationTestBase):
def setUpSql(self):
self.sql_file(os.path.join(_THIS_DIR, 'data', 'create.sql'))
self.sql_file(os.path.join(_THIS_DIR, 'data', 'insert.sql'))
def test_execute_type_blob(self):
cursor = self.conn.cursor()
stmt = "insert into ACCOUNT (ACCOUNT_ID, ACCOUNT_NO, BALANCE, " \
"STUFF) values (?, ?, ?, ?)"
stuff = self.dbapi.Binary('abcdef')
parms = ('2009-09-11 14:15:22.123450', 20, 13.1, stuff)
cursor.execute(stmt, parms)
stmt = "select STUFF from ACCOUNT where ACCOUNT_NO = ?"
parms = (20, )
cursor.execute(stmt, parms)
result = cursor.fetchone()
cursor.close()
value = result[0]
self.assertEqual(value, buffer('abcdef'))
@unittest.skipIf(is_jython(), "requires python")
class SqlitePyTest(SqliteTestBase, unittest.TestCase):
def connect(self):
import sqlite3
return sqlite3, sqlite3.connect(':memory:')
class SqliteXerialTest(SqliteTestBase, unittest.TestCase):
def connect(self):
#http://bitbucket.org/xerial/sqlite-jdbc
# sqlite-jdbc-3.7.2.jar
driver, driver_args = 'org.sqlite.JDBC', 'jdbc:sqlite::memory:'
# db2jcc
# driver, driver_args = 'com.ibm.db2.jcc.DB2Driver', \
# ['jdbc:db2://4.100.73.81:50000/db2t', 'user', 'passwd']
# driver from http://www.ch-werner.de/javasqlite/ seems to be
# crap as it returns decimal values as VARCHAR type
# sqlite.jar
# driver, driver_args = 'SQLite.JDBCDriver', 'jdbc:sqlite:/:memory:'
# Oracle Thin Driver
# driver, driver_args = 'oracle.jdbc.OracleDriver', \
# ['jdbc:oracle:thin:@//hh-cluster-scan:1521/HH_TPP',
# 'user', 'passwd']
return jaydebeapi, jaydebeapi.connect(driver, driver_args)
@unittest.skipUnless(is_jython(), "don't know how to support blob")
def test_execute_type_blob(self):
return super(SqliteXerialTest, self).test_execute_type_blob()
class HsqldbTest(IntegrationTestBase, unittest.TestCase):
def connect(self):
# http://hsqldb.org/
# hsqldb.jar
driver, driver_args = 'org.hsqldb.jdbcDriver', ['jdbc:hsqldb:mem:.',
'SA', '']
return jaydebeapi, jaydebeapi.connect(driver, driver_args)
def setUpSql(self):
self.sql_file(os.path.join(_THIS_DIR, 'data', 'create_hsqldb.sql'))
self.sql_file(os.path.join(_THIS_DIR, 'data', 'insert.sql'))