importmath
importthreading
importpytest
fromsqlobjectimport SQLObject, StringCol, FloatCol
fromsqlobject.compatimport string_type
fromsqlobject.sqlbuilderimport Select
fromsqlobject.tests.dbtestimport getConnection, setupClass, supports
fromsqlobject.tests.dbtestimport setSQLiteConnectionFactory
from.test_basicimport SOTestSO1
try:
connection = getConnection()
except (AttributeError, NameError):
# The module was imported during documentation building
pass
else:
if connection.dbName != "sqlite":
pytestmark = pytest.mark.skip("These tests require SQLite")
[docs]
classSQLiteFactoryTest(SQLObject):
name = StringCol()
[docs]
deftest_sqlite_factory():
setupClass(SQLiteFactoryTest)
factory = [None]
defSQLiteConnectionFactory(sqlite):
classMyConnection(sqlite.Connection):
pass
factory[0] = MyConnection
return MyConnection
setSQLiteConnectionFactory(SQLiteFactoryTest, SQLiteConnectionFactory)
conn = SQLiteFactoryTest._connection.makeConnection()
assert factory[0]
assert isinstance(conn, factory[0])
[docs]
deftest_sqlite_factory_str():
setupClass(SQLiteFactoryTest)
factory = [None]
defSQLiteConnectionFactory(sqlite):
classMyConnection(sqlite.Connection):
pass
factory[0] = MyConnection
return MyConnection
fromsqlobject.sqliteimport sqliteconnection
sqliteconnection.SQLiteConnectionFactory = SQLiteConnectionFactory
setSQLiteConnectionFactory(SQLiteFactoryTest, "SQLiteConnectionFactory")
conn = SQLiteFactoryTest._connection.makeConnection()
assert factory[0]
assert isinstance(conn, factory[0])
del sqliteconnection.SQLiteConnectionFactory
[docs]
deftest_sqlite_aggregate():
setupClass(SQLiteFactoryTest)
defSQLiteConnectionFactory(sqlite):
classMyConnection(sqlite.Connection):
def__init__(self, *args, **kwargs):
super(MyConnection, self).__init__(*args, **kwargs)
self.create_aggregate("group_concat", 1, self.group_concat)
classgroup_concat:
def__init__(self):
self.acc = []
defstep(self, value):
if isinstance(value, string_type):
self.acc.append(value)
else:
self.acc.append(str(value))
deffinalize(self):
self.acc.sort()
return ", ".join(self.acc)
return MyConnection
setSQLiteConnectionFactory(SQLiteFactoryTest, SQLiteConnectionFactory)
SQLiteFactoryTest(name='sqlobject')
SQLiteFactoryTest(name='sqlbuilder')
assert SQLiteFactoryTest.select(orderBy="name").\
accumulateOne("group_concat", "name") == \
"sqlbuilder, sqlobject"
[docs]
defdo_select():
list(SOTestSO1.select())
[docs]
deftest_sqlite_threaded():
setupClass(SOTestSO1)
t = threading.Thread(target=do_select)
t.start()
t.join()
# This should reuse the same connection as the connection
# made above (at least will with most database drivers, but
# this will cause an error in SQLite):
do_select()
[docs]
deftest_empty_string():
setupClass(SOTestSO1)
test = SOTestSO1(name=None, passwd='')
assert test.name is None
assert test.passwd == ''
[docs]
deftest_memorydb():
if not supports("memorydb"):
pytest.skip("memorydb isn't supported")
if not connection._memory:
pytest.skip("The connection isn't memorydb")
setupClass(SOTestSO1)
connection.close() # create a new connection to an in-memory database
SOTestSO1.setConnection(connection)
SOTestSO1.createTable()
[docs]
deftest_list_databases():
assert 'main' in connection.listDatabases()
[docs]
deftest_list_tables():
setupClass(SOTestSO1)
assert SOTestSO1.sqlmeta.table in connection.listTables()
[docs]
classSQLiteTruedivTest(SQLObject):
value = FloatCol()
[docs]
deftest_truediv():
setupClass(SQLiteTruedivTest)
if SQLiteTruedivTest._connection.dbName == "sqlite":
defSQLiteConnectionFactory(sqlite):
classMyConnection(sqlite.Connection):
def__init__(self, *args, **kwargs):
super(MyConnection, self).__init__(*args, **kwargs)
self.create_function("floor", 1, math.floor)
return MyConnection
setSQLiteConnectionFactory(SQLiteTruedivTest, SQLiteConnectionFactory)
SQLiteTruedivTest(value=-5.0)
assert SQLiteTruedivTest._connection.queryAll(
SQLiteTruedivTest._connection.sqlrepr(
Select(SQLiteTruedivTest.q.value // 4)))[0][0] == -2