|
From: <sub...@co...> - 2006-10-10 12:03:16
|
Author: phd
Date: 2006-10-10 06:03:10 -0600 (Tue, 10 Oct 2006)
New Revision: 1992
Added:
SQLObject/trunk/sqlobject/tests/test_sqlite_factory.py
Modified:
SQLObject/trunk/sqlobject/sqlite/sqliteconnection.py
SQLObject/trunk/sqlobject/tests/test_select.py
Log:
SQLiteConnection got a new keyword "factory". The patch inspired by jt <jt...@ud...>.
Modified: SQLObject/trunk/sqlobject/sqlite/sqliteconnection.py
===================================================================
--- SQLObject/trunk/sqlobject/sqlite/sqliteconnection.py 2006-10-10 04:19:44 UTC (rev 1991)
+++ SQLObject/trunk/sqlobject/sqlite/sqliteconnection.py 2006-10-10 12:03:10 UTC (rev 1992)
@@ -67,6 +67,11 @@
if sqlite2_Binary is None:
sqlite2_Binary = sqlite.Binary
sqlite.Binary = lambda s: sqlite2_Binary(sqlite.encode(s))
+ if 'factory' in kw:
+ factory = popKey(kw, 'factory')
+ if isinstance(factory, str):
+ factory = globals()[factory]
+ opts['factory'] = factory(sqlite)
else:
opts['autocommit'] = bool(autoCommit)
if 'encoding' in kw:
Modified: SQLObject/trunk/sqlobject/tests/test_select.py
===================================================================
--- SQLObject/trunk/sqlobject/tests/test_select.py 2006-10-10 04:19:44 UTC (rev 1991)
+++ SQLObject/trunk/sqlobject/tests/test_select.py 2006-10-10 12:03:10 UTC (rev 1992)
@@ -2,6 +2,7 @@
from sqlobject import *
from sqlobject.main import SQLObjectIntegrityError
from sqlobject.tests.dbtest import *
+from sqlobject.tests.dbtest import installOrClear
from py.test import raises
try:
@@ -133,7 +134,7 @@
def test_2(self):
self.accumulateEqual(len,Counter2.select('all'), 100)
-def test_sqlbuilder_LIKE():
+def test_select_LIKE():
setupClass(IterTest)
IterTest(name='sqlobject')
IterTest(name='sqlbuilder')
@@ -142,12 +143,12 @@
assert IterTest.select(LIKE(IterTest.q.name, "sqlb%")).count() == 1
assert IterTest.select(LIKE(IterTest.q.name, "sqlx%")).count() == 0
-def test_sqlbuilder_RLIKE():
+def test_select_RLIKE():
setupClass(IterTest)
if IterTest._connection.dbName == "sqlite":
- from sqlobject.sqlite.sqliteconnection import using_sqlite2
- if not using_sqlite2:
+ from sqlobject.sqlite import sqliteconnection
+ if not sqliteconnection.using_sqlite2:
return
# Implement regexp() function for SQLite; only works with PySQLite2
@@ -155,13 +156,23 @@
def regexp(regexp, test):
return bool(re.search(regexp, test))
- _get_connection = IterTest._connection.getConnection
- def new_get_connection(*args, **kw):
- _connection = _get_connection(*args, **kw)
- _connection.create_function("regexp", 2, regexp)
- return _connection
- IterTest._connection.getConnection = new_get_connection
+ def SQLiteConnectionFactory(sqlite):
+ class MyConnection(sqlite.Connection):
+ def __init__(self, *args, **kwargs):
+ super(MyConnection, self).__init__(*args, **kwargs)
+ self.create_function("regexp", 2, regexp)
+ return MyConnection
+ conn = IterTest._connection
+ IterTest._connection = sqliteconnection.SQLiteConnection(
+ filename=conn.filename,
+ name=conn.name, debug=conn.debug, debugOutput=conn.debugOutput,
+ cache=conn.cache, style=conn.style, autoCommit=conn.autoCommit,
+ debugThreading=conn.debugThreading, registry=conn.registry,
+ factory=SQLiteConnectionFactory
+ )
+ installOrClear([IterTest])
+
IterTest(name='sqlobject')
IterTest(name='sqlbuilder')
assert IterTest.select(RLIKE(IterTest.q.name, "^sql.*$")).count() == 2
Added: SQLObject/trunk/sqlobject/tests/test_sqlite_factory.py
===================================================================
--- SQLObject/trunk/sqlobject/tests/test_sqlite_factory.py (rev 0)
+++ SQLObject/trunk/sqlobject/tests/test_sqlite_factory.py 2006-10-10 12:03:10 UTC (rev 1992)
@@ -0,0 +1,105 @@
+from sqlobject import *
+from sqlobject.tests.dbtest import *
+from sqlobject.tests.dbtest import installOrClear
+
+class SQLiteFactoryTest(SQLObject):
+ name = StringCol()
+
+def test_sqlite_factory():
+ setupClass(SQLiteFactoryTest)
+
+ if SQLiteFactoryTest._connection.dbName == "sqlite":
+ from sqlobject.sqlite import sqliteconnection
+ if not sqliteconnection.using_sqlite2:
+ return
+
+ factory = [None]
+ def SQLiteConnectionFactory(sqlite):
+ class MyConnection(sqlite.Connection):
+ pass
+ factory[0] = MyConnection
+ return MyConnection
+
+ conn = SQLiteFactoryTest._connection
+ SQLiteFactoryTest._connection = sqliteconnection.SQLiteConnection(
+ filename=conn.filename,
+ name=conn.name, debug=conn.debug, debugOutput=conn.debugOutput,
+ cache=conn.cache, style=conn.style, autoCommit=conn.autoCommit,
+ debugThreading=conn.debugThreading, registry=conn.registry,
+ factory=SQLiteConnectionFactory
+ )
+ conn = SQLiteFactoryTest._connection.makeConnection()
+ assert factory[0]
+ assert isinstance(conn, factory[0])
+
+def test_sqlite_factory_str():
+ setupClass(SQLiteFactoryTest)
+
+ if SQLiteFactoryTest._connection.dbName == "sqlite":
+ from sqlobject.sqlite import sqliteconnection
+ if not sqliteconnection.using_sqlite2:
+ return
+
+ factory = [None]
+ def SQLiteConnectionFactory(sqlite):
+ class MyConnection(sqlite.Connection):
+ pass
+ factory[0] = MyConnection
+ return MyConnection
+ sqliteconnection.SQLiteConnectionFactory = SQLiteConnectionFactory
+
+ conn = SQLiteFactoryTest._connection
+ SQLiteFactoryTest._connection = sqliteconnection.SQLiteConnection(
+ filename=conn.filename,
+ name=conn.name, debug=conn.debug, debugOutput=conn.debugOutput,
+ cache=conn.cache, style=conn.style, autoCommit=conn.autoCommit,
+ debugThreading=conn.debugThreading, registry=conn.registry,
+ factory="SQLiteConnectionFactory"
+ )
+ conn = SQLiteFactoryTest._connection.makeConnection()
+ assert factory[0]
+ assert isinstance(conn, factory[0])
+ del sqliteconnection.SQLiteConnectionFactory
+
+def test_sqlite_aggregate():
+ setupClass(SQLiteFactoryTest)
+
+ if SQLiteFactoryTest._connection.dbName == "sqlite":
+ from sqlobject.sqlite import sqliteconnection
+ if not sqliteconnection.using_sqlite2:
+ return
+
+ def SQLiteConnectionFactory(sqlite):
+ class MyConnection(sqlite.Connection):
+ def __init__(self, *args, **kwargs):
+ super(MyConnection, self).__init__(*args, **kwargs)
+ self.create_aggregate("group_concat", 1, self.group_concat)
+
+ class group_concat:
+ def __init__(self):
+ self.acc = []
+ def step(self, value):
+ if isinstance(value, basestring):
+ self.acc.append(value)
+ else:
+ self.acc.append(str(value))
+ def finalize(self):
+ self.acc.sort()
+ return ", ".join(self.acc)
+
+ return MyConnection
+
+ conn = SQLiteFactoryTest._connection
+ SQLiteFactoryTest._connection = sqliteconnection.SQLiteConnection(
+ filename=conn.filename,
+ name=conn.name, debug=conn.debug, debugOutput=conn.debugOutput,
+ cache=conn.cache, style=conn.style, autoCommit=conn.autoCommit,
+ debugThreading=conn.debugThreading, registry=conn.registry,
+ factory=SQLiteConnectionFactory
+ )
+ installOrClear([SQLiteFactoryTest])
+
+ SQLiteFactoryTest(name='sqlobject')
+ SQLiteFactoryTest(name='sqlbuilder')
+ assert SQLiteFactoryTest.select(orderBy="name").accumulateOne("group_concat", "name") == \
+ "sqlbuilder, sqlobject"
|
