[Zope3-checkins] SVN: Zope3/trunk/ Fixed issue #445: gadflyDA not
full implementation
Dmitry Vasiliev
dima at hlabs.spb.ru
Mon Aug 29 10:29:58 EDT 2005
Log message for revision 38136:
Fixed issue #445: gadflyDA not full implementation
Changed:
U Zope3/trunk/doc/CHANGES.txt
U Zope3/trunk/src/zope/app/rdb/__init__.py
U Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py
U Zope3/trunk/src/zope/app/rdb/tests/test_zopecursor.py
-=-
Modified: Zope3/trunk/doc/CHANGES.txt
===================================================================
--- Zope3/trunk/doc/CHANGES.txt 2005-08-29 13:53:57 UTC (rev 38135)
+++ Zope3/trunk/doc/CHANGES.txt 2005-08-29 14:29:57 UTC (rev 38136)
@@ -32,6 +32,8 @@
Bug Fixes
+ - Fixed issue #445: gadflyDA not full implementation
+
- Fix: Pagelet directive was inoring the permission. Pagelets which
not the right permission are not shwon if calling via the Tales
expression ``pagelets:``. If calling named pagelets directly via the
Modified: Zope3/trunk/src/zope/app/rdb/__init__.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/__init__.py 2005-08-29 13:53:57 UTC (rev 38135)
+++ Zope3/trunk/src/zope/app/rdb/__init__.py 2005-08-29 14:29:57 UTC (rev 38136)
@@ -290,21 +290,28 @@
return self.cursor.execute(operation)
return self.cursor.execute(operation, parameters)
+ def executemany(self, operation, parameters):
+ """Executes an operation, registering the underlying
+ connection with the transaction system. """
+ operation, parameters = self._prepareOperation(operation, parameters)
+ method = getattr(self.cursor, "executemany", self.cursor.execute)
+ self.connection.registerForTxn()
+ return method(operation, parameters)
+
def _prepareOperation(self, operation, parameters):
encoding = self.connection.getTypeInfo().getEncoding()
if isinstance(operation, unicode):
operation = operation.encode(encoding)
+ parameters = self._prepareParameters(parameters, encoding)
+ return operation, parameters
+ def _prepareParameters(self, parameters, encoding):
if isinstance(parameters, list):
for i, v in enumerate(parameters):
if isinstance(v, unicode):
parameters[i] = v.encode(encoding)
- elif isinstance(v, tuple):
- values = list(v)
- for j, v in enumerate(values):
- if isinstance(v, unicode):
- values[j] = v.encode(encoding)
- parameters[i] = tuple(values)
+ else:
+ parameters[i] = self._prepareParameters(v, encoding)
elif isinstance(parameters, tuple):
parameters = list(parameters)
for i, v in enumerate(parameters):
@@ -315,7 +322,7 @@
for k, v in parameters.items():
if isinstance(v, unicode):
parameters[k] = v.encode(encoding)
- return operation, parameters
+ return parameters
def __getattr__(self, key):
return getattr(self.cursor, key)
Modified: Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py 2005-08-29 13:53:57 UTC (rev 38135)
+++ Zope3/trunk/src/zope/app/rdb/tests/test_gadflyadapter.py 2005-08-29 14:29:57 UTC (rev 38136)
@@ -19,9 +19,13 @@
import tempfile
from unittest import TestCase, TestSuite, main, makeSuite
+from zope.interface.verify import verifyObject
+
from zope.app.rdb import DatabaseAdapterError
+from zope.app.rdb.interfaces import IZopeConnection, IZopeCursor
from zope.app.rdb.gadflyda import GadflyAdapter, setGadflyRoot
+
class GadflyTestBase(TestCase):
def setUp(self):
@@ -42,8 +46,7 @@
return self.tempdir
def _create(self, *args):
- obj = GadflyAdapter(*args)
- return obj
+ return GadflyAdapter(*args)
class TestGadflyAdapter(GadflyTestBase):
@@ -131,7 +134,14 @@
conn = a._connection_factory()
conn.rollback() # is it really a connection?
+ def test__interfaces(self):
+ a = self._create("dbi://demo")
+ connection = a()
+ verifyObject(IZopeConnection, connection)
+ cursor = connection.cursor()
+ verifyObject(IZopeCursor, cursor)
+
def test_suite():
return TestSuite((
makeSuite(TestGadflyAdapter),
Modified: Zope3/trunk/src/zope/app/rdb/tests/test_zopecursor.py
===================================================================
--- Zope3/trunk/src/zope/app/rdb/tests/test_zopecursor.py 2005-08-29 13:53:57 UTC (rev 38135)
+++ Zope3/trunk/src/zope/app/rdb/tests/test_zopecursor.py 2005-08-29 14:29:57 UTC (rev 38136)
@@ -25,7 +25,11 @@
def cursor(self):
return MyCursorStub()
+class MyConnectionStub2(ConnectionStub):
+ def cursor(self):
+ return MyCursorStub2()
+
raw = [['mano', 2, 'buvo batai'],
['dingo', 1, 'nerandu'],
['as su', 1, 'batuku'],
@@ -44,6 +48,7 @@
description = ((None, 'string'), (None, 'int'), (None, 'foo'))
def execute(self, query, args=None):
+ self.method = "execute"
self.query = query
self.args = args
@@ -59,7 +64,14 @@
def fetchmany(self, size=2):
return self._raw[:size]
+class MyCursorStub2(MyCursorStub):
+ def executemany(self, query, args):
+ self.method = "executemany"
+ self.query = query
+ self.args = args
+
+
class MyTypeInfoStub(TypeInfoStub):
def getConverter(self, type):
@@ -125,6 +137,19 @@
'got %r,\n'
'expected %r' % (results, expected))
+ def test_cursor_executemany(self):
+ self.cursor.executemany("SELECT", [("A",), ("B",)])
+ self.assertEqual("execute", self.cursor.cursor.method)
+ self.assertEqual("SELECT", self.cursor.cursor.query)
+ self.assertEqual([("A",), ("B",)], self.cursor.cursor.args)
+
+ zc = ZopeConnection(MyConnectionStub2(), self.typeInfo)
+ self.cursor = ZopeCursor(zc.conn.cursor(), zc)
+ self.cursor.executemany("SELECT", [("A",), ("B",)])
+ self.assertEqual("executemany", self.cursor.cursor.method)
+ self.assertEqual("SELECT", self.cursor.cursor.query)
+ self.assertEqual([("A",), ("B",)], self.cursor.cursor.args)
+
def test_cursor_query_encoding(self):
self.cursor.execute(u'\u0422\u0435\u0441\u0442')
self.assertEqual('\xd0\xa2\xd0\xb5\xd1\x81\xd1\x82',
@@ -151,6 +176,15 @@
[(u'\u0422\u0435\u0441\u0442',)])
self.assertEqual([('\xd2\xe5\xf1\xf2',)], self.cursor.cursor.args)
+ self.cursor.execute("SELECT * FROM table",
+ [[u'\u0422\u0435\u0441\u0442']])
+ self.assertEqual([['\xd2\xe5\xf1\xf2']], self.cursor.cursor.args)
+
+ self.cursor.execute("SELECT * FROM table",
+ [{"value": u'\u0422\u0435\u0441\u0442'}])
+ self.assertEqual([{"value": '\xd2\xe5\xf1\xf2'}],
+ self.cursor.cursor.args)
+
def test_cursor_dict_args_encoding(self):
self.typeInfo.setEncoding("windows-1251")
self.cursor.execute("SELECT * FROM table",
More information about the Zope3-Checkins
mailing list