[Zope3-checkins] CVS: Zope3/src/zope/app/rdb/tests - __init__.py:1.2 stubs.py:1.2 test_dsnparser.py:1.2 test_gadflyadapter.py:1.2 test_resultset.py:1.2 test_row.py:1.2 test_sqlcommand.py:1.2 test_zopeconnection.py:1.2 test_zopecursor.py:1.2 test_zopedatabaseadapter.py:1.2 test_zopedbtransactionmanager.py:1.2
Jim Fulton
jim@zope.com
Wed, 25 Dec 2002 09:13:46 -0500
Update of /cvs-repository/Zope3/src/zope/app/rdb/tests
In directory cvs.zope.org:/tmp/cvs-serv15352/src/zope/app/rdb/tests
Added Files:
__init__.py stubs.py test_dsnparser.py test_gadflyadapter.py
test_resultset.py test_row.py test_sqlcommand.py
test_zopeconnection.py test_zopecursor.py
test_zopedatabaseadapter.py test_zopedbtransactionmanager.py
Log Message:
Grand renaming:
- Renamed most files (especially python modules) to lower case.
- Moved views and interfaces into separate hierarchies within each
project, where each top-level directory under the zope package
is a separate project.
- Moved everything to src from lib/python.
lib/python will eventually go away. I need access to the cvs
repository to make this happen, however.
There are probably some bits that are broken. All tests pass
and zope runs, but I haven't tried everything. There are a number
of cleanups I'll work on tomorrow.
=== Zope3/src/zope/app/rdb/tests/__init__.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/__init__.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,2 @@
+#
+# This file is necessary to make this directory a package.
=== Zope3/src/zope/app/rdb/tests/stubs.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/stubs.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,47 @@
+##############################################################################
+#
+# Copyright (c) 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Stubs for Zope RDB unit tests.
+
+$Id$
+"""
+
+class ConnectionStub:
+
+ def __init__(self):
+ self._called={}
+
+ def cursor(self):
+ return CursorStub()
+
+ def answer(self):
+ return 42
+
+ def commit(self, *ignored):
+ v = self._called.setdefault('commit',0)
+ v+=1
+ self._called['commit']=v
+ def rollback(self, *ignored):
+ v = self._called.setdefault('rollback',0)
+ v+=1
+ self._called['rollback']=v
+
+class CursorStub:
+ def execute(*args, **kw):
+ pass
+
+class TypeInfoStub:
+ paramstyle = 'pyformat'
+ threadsafety = 0
+ def getConverter(self, type):
+ return lambda x: x
=== Zope3/src/zope/app/rdb/tests/test_dsnparser.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_dsnparser.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,72 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+import unittest
+from zope.app.rdb import parseDSN
+
+
+class TestDSNParser(unittest.TestCase):
+
+ def testDBNameOnly(self):
+ dsn = 'dbi://test'
+ result = {'parameters': {}, 'dbname': 'test', 'username': '',
+ 'password': '', 'host': '', 'port': ''}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testDBNameAndParams(self):
+ dsn = 'dbi://test;param1=value1;param2=value2'
+ result = {'parameters': {'param1': 'value1', 'param2': 'value2'},
+ 'dbname': 'test', 'username': '', 'password': '',
+ 'host': '', 'port': ''}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testUserPassword(self):
+ dsn = 'dbi://mike:muster/test'
+ result = {'parameters': {}, 'dbname': 'test', 'username': 'mike',
+ 'password': 'muster', 'host': '', 'port': ''}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testUserPasswordAndParams(self):
+ dsn = 'dbi://mike:muster/test;param1=value1;param2=value2'
+ result = {'parameters': {'param1': 'value1', 'param2': 'value2'},
+ 'dbname': 'test', 'username': 'mike', 'password': 'muster',
+ 'host': '', 'port': ''}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testAllOptions(self):
+ dsn = 'dbi://mike:muster@bohr:5432/test'
+ result = {'parameters': {}, 'dbname': 'test', 'username': 'mike',
+ 'password': 'muster', 'host': 'bohr', 'port': '5432'}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testAllOptionsAndParams(self):
+ dsn = 'dbi://mike:muster@bohr:5432/test;param1=value1;param2=value2'
+ result = {'parameters': {'param1': 'value1', 'param2': 'value2'},
+ 'dbname': 'test', 'username': 'mike', 'password': 'muster',
+ 'host': 'bohr', 'port': '5432'}
+ self.assertEqual(result, parseDSN(dsn))
+
+ def testFailures(self):
+ self.assertRaises(AssertionError, parseDSN, None)
+ self.assertRaises(AssertionError, parseDSN, 'dfi://')
+
+def test_suite():
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.makeSuite(TestDSNParser))
+ return suite
+
+if __name__ == '__main__':
+ unittest.TextTestRunner().run(test_suite())
=== Zope3/src/zope/app/rdb/tests/test_gadflyadapter.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_gadflyadapter.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,159 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Gadfly database adapter unit tests.
+
+$Id$
+"""
+
+import os
+import tempfile
+from unittest import TestCase, TestSuite, main, makeSuite
+from zope.app.rdb import DatabaseAdapterError
+
+try:
+ from tempfile import mkdtemp
+except ImportError:
+ import errno
+
+ def mkdtemp(suffix=""):
+ """Poor man's version of tempfile.mkdtemp from Python 2.3"""
+
+ for seq in xrange(1000):
+ name = tempfile.mktemp(suffix)
+ try:
+ os.mkdir(name, 0700)
+ return name
+ except IOError, e:
+ if e.errno == errno.EEXIST:
+ continue
+ raise
+
+ raise IOError(errno.EEXIST, "No usable temporary directory name found")
+
+
+class GadflyTestBase(TestCase):
+
+ def setUp(self):
+ TestCase.setUp(self)
+ self.tempdir = None
+
+ def tearDown(self):
+ TestCase.tearDown(self)
+ if self.tempdir:
+ os.rmdir(self.tempdir)
+
+ def getGadflyRoot(self):
+ # note that self is GadflyTestBase here
+ if not self.tempdir:
+ self.tempdir = mkdtemp('gadfly')
+ return self.tempdir
+
+ def _create(self, *args):
+ from zope.app.rdb.gadflyda import GadflyAdapter
+ obj = GadflyAdapter(*args)
+ obj._getGadflyRoot = self.getGadflyRoot
+ return obj
+
+
+class TestGadflyAdapter(GadflyTestBase):
+ """Test incorrect connection strings"""
+
+ def test__connection_factory_nonexistent(self):
+ # Should raise an exception on nonexistent dirs.
+ a = self._create("dbi://demo;dir=nonexistent")
+ self.assertRaises(DatabaseAdapterError, a._connection_factory)
+
+ def test__connection_factory_bad_dsn(self):
+ a = self._create("dbi://user:pass/demo;dir=nonexistent")
+ self.assertRaises(DatabaseAdapterError, a._connection_factory)
+
+ a = self._create("dbi://localhost:1234/demo;dir=nonexistent")
+ self.assertRaises(DatabaseAdapterError, a._connection_factory)
+
+
+class TestGadflyAdapterNew(GadflyTestBase):
+ """Test with nonexistent databases"""
+
+ def test__connection_factory_create(self):
+ # Should create a database if the directory is empty.
+ a = self._create("dbi://demo;dir=test")
+ conn = a._connection_factory()
+ conn.rollback() # is it really a connection?
+
+ def test__connection_factory_existing(self):
+ # Should fail gracefully if the directory is a file.
+ open(os.path.join(self.getGadflyRoot(), "regular"), "w").close()
+ a = self._create("dbi://demo;dir=regular")
+ self.assertRaises(DatabaseAdapterError, a._connection_factory)
+
+ def setUp(self):
+ # Create a directory for the database.
+ GadflyTestBase.setUp(self)
+ dir = self.getGadflyRoot()
+ os.mkdir(os.path.join(dir, "test"))
+
+ def tearDown(self):
+ # Remove the files and directories created.
+ dir = self.getGadflyRoot()
+ try: os.unlink(os.path.join(dir, "test", "demo.gfd"))
+ except: pass
+ os.rmdir(os.path.join(dir, "test"))
+ try: os.unlink(os.path.join(dir, "regular"))
+ except: pass
+ GadflyTestBase.tearDown(self)
+
+
+class TestGadflyAdapterDefault(GadflyTestBase):
+ """Test with pre-existing databases"""
+
+ def test__connection_factory_create(self):
+ # Should create a database if the directory is empty.
+ a = self._create("dbi://demo")
+ conn = a._connection_factory()
+ conn.rollback() # is it really a connection?
+
+ def test__connection_factory_reopen(self):
+ # Should open an existing database.
+ a = self._create("dbi://demo")
+ conn = a._connection_factory()
+ conn.rollback() # is it really a connection?
+ conn.close()
+
+ conn = a._connection_factory()
+ conn.rollback() # is it really a connection?
+
+ def setUp(self):
+ # Create a directory for the database.
+ GadflyTestBase.setUp(self)
+ dir = self.getGadflyRoot()
+ os.mkdir(os.path.join(dir, "demo"))
+
+ def tearDown(self):
+ # Remove the files and directories created.
+ dir = self.getGadflyRoot()
+ try: os.unlink(os.path.join(dir, "demo", "demo.gfd"))
+ except: pass
+ os.rmdir(os.path.join(dir, "demo"))
+ GadflyTestBase.tearDown(self)
+
+
+def test_suite():
+ return TestSuite((
+ makeSuite(TestGadflyAdapter),
+ makeSuite(TestGadflyAdapterNew),
+ makeSuite(TestGadflyAdapterDefault),
+ ))
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_resultset.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_resultset.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,73 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""ResultSet unit tests.
+
+$Id$
+"""
+
+from unittest import TestCase, TestSuite, main, makeSuite
+
+class TestResultSet(TestCase):
+
+ def testPickling(self):
+ from zope.app.rdb import ResultSet
+ from pickle import dumps, loads
+
+ columns = ('foo', 'bar')
+ rows = (('1', '2'), ('3', '4'))
+ rs = ResultSet(columns, rows)
+
+ pickled = dumps(rs)
+ unpickled = loads(pickled)
+
+ #self.assertEqual(unpickled.columns, rs.columns)
+ self.assertEqual(rs, unpickled)
+
+ def test__cmp__(self):
+ from zope.app.rdb import ResultSet
+ from copy import deepcopy
+
+ # See if equal to a copy
+ columns = ('foo', 'bar')
+ rows = (('1', '2'), ('3', '4'))
+ rs1 = ResultSet(columns, rows)
+ rs2 = ResultSet(deepcopy(columns), deepcopy(rows))
+ self.assertEqual(rs1, rs2, "deep copy not equal")
+ self.assertEqual(rs1, rs1, "not equal to self")
+
+ # Test if the columns are different
+ columns1 = ('foo', 'bar')
+ rows = (('1', '2'), ('3', '4'))
+ rs1 = ResultSet(columns1, rows)
+ columns2 = ('Foo', 'Bar')
+ rs2 = ResultSet(columns2, rows)
+ self.assert_(rs1 > rs2, "different columns compared incorrectly")
+
+ # Test if the data is different
+ columns = ('foo', 'bar')
+ rows1 = (('1', '2'), ('3', '4'))
+ rows2 = (('2', '2'), ('3', '4'))
+ rs1 = ResultSet(columns, rows1)
+ rs2 = ResultSet(columns, rows2)
+ self.assert_(rs1 < rs2, "different columns compared incorrectly")
+
+
+
+def test_suite():
+ return TestSuite((
+ makeSuite(TestResultSet),
+ ))
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_row.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_row.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,90 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Row class tests.
+
+$Id$
+"""
+
+from unittest import TestCase, TestSuite, main, makeSuite
+
+class RowTests(TestCase):
+
+ def test_RowClassFactory(self):
+ from zope.app.rdb import RowClassFactory
+
+ columns = ('food', 'name')
+ data = ('pizza', 'john')
+
+ klass = RowClassFactory(columns)
+ ob = klass(data)
+
+ self.failUnless(ob.food == 'pizza', "bad row class attribute")
+ self.failUnless(ob.name == 'john', "bad row class attribute (2)")
+
+ def test_RowClassFactory_Proxied(self):
+ from zope.app.rdb import RowClassFactory
+ from zope.security.proxy import ProxyFactory
+ from zope.exceptions import ForbiddenAttribute
+
+ columns = ('type', 'speed')
+ data = ('airplane', '800km')
+
+ klass = RowClassFactory(columns)
+
+ ob = klass(data)
+
+ proxied = ProxyFactory(ob)
+
+ self.failUnless (proxied.type == 'airplane', "security proxy error")
+ self.failUnless (proxied.speed == '800km', "security proxy error (2)")
+
+ self.assertRaises(ForbiddenAttribute,
+ lambda x=proxied: x.__slots__
+ )
+
+ def test__cmp__(self):
+ from zope.app.rdb import RowClassFactory
+
+ columns = ('food', 'name')
+ data = ('pizza', 'john')
+
+ klass = RowClassFactory(columns)
+ ob = klass(data)
+ self.assertEqual(ob, ob, "not equal to self")
+
+ klass2 = RowClassFactory(columns)
+ ob2 = klass2(data)
+ self.assertEqual(ob, ob2, "not equal to an identical class")
+
+ columns = ('food', 'surname')
+ data = ('pizza', 'john')
+
+ klass3 = RowClassFactory(columns)
+ ob3 = klass3(data)
+ self.assert_(ob < ob3, "cmp with different columns")
+
+ columns = ('food', 'name')
+ data = ('pizza', 'mary')
+
+ klass4 = RowClassFactory(columns)
+ ob4 = klass4(data)
+ self.assert_(ob < ob4, "cmp with different data")
+
+
+
+def test_suite():
+ return makeSuite(RowTests)
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_sqlcommand.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_sqlcommand.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,85 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+
+import unittest
+
+from zope.app.component import nextservice
+from zope.app.interfaces.rdb import IConnectionService
+from zope.app.interfaces.rdb import IZopeConnection
+from zope.app.interfaces.rdb import IZopeCursor
+from zope.app.rdb import SQLCommand
+from zope.app.tests.placelesssetup import PlacelessSetup
+from zope.component.service import serviceManager as sm
+
+
+# Make some fixes, so that we overcome some of the natural ZODB properties
+def getNextServiceManager(context):
+ return sm
+
+class CursorStub:
+
+ __implements__ = IZopeCursor
+
+ description = (('id', 'int'),)
+
+ def execute(self, operation, parameters=None):
+ self.result = {"SELECT id FROM Table": ((1,),)}[operation]
+
+ def fetchall(self):
+ return self.result
+
+
+class ConnectionStub:
+
+ __implements__ = IZopeConnection
+
+ def cursor(self):
+ return CursorStub()
+
+
+class ConnectionServiceStub:
+
+ __implements__ = IConnectionService
+
+ def getConnection(self, name):
+ return ConnectionStub()
+
+
+class SQLCommandTest(unittest.TestCase, PlacelessSetup):
+
+ def setUp(self):
+ PlacelessSetup.setUp(self)
+ sm.defineService('SQLDatabaseConnections', IConnectionService)
+ sm.provideService('SQLDatabaseConnections', ConnectionServiceStub())
+ self._old_getNextServiceManager = nextservice.getNextServiceManager
+ nextservice.getNextServiceManager = getNextServiceManager
+
+ def tearDown(self):
+ nextservice.getNextServiceManager = self._old_getNextServiceManager
+
+ def testsimp(self):
+ command = SQLCommand("my_connection", "SELECT id FROM Table")
+ result = command()
+ self.assertEqual(result.columns, ('id',))
+ self.assertEqual(result[0].id, 1)
+
+
+def test_suite():
+ return unittest.makeSuite(SQLCommandTest)
+
+if __name__=='__main__':
+ unittest.main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_zopeconnection.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_zopeconnection.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,76 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+
+from unittest import TestCase, TestSuite, main, makeSuite
+from transaction import get_transaction
+from zope.app.rdb import ZopeConnection
+from zope.app.interfaces.rdb import IZopeCursor
+from zope.app.rdb.tests.stubs import ConnectionStub, TypeInfoStub
+
+class ZopeConnectionTests(TestCase):
+
+ def test_cursor(self):
+ zc = ZopeConnection(ConnectionStub(), TypeInfoStub())
+ cursor = zc.cursor()
+
+ self.failUnless(IZopeCursor.isImplementedBy(cursor),
+ "cursor is not what we expected")
+
+ def test_connection_txn_registration(self):
+ t = get_transaction()
+ t.begin()
+
+ zc = ZopeConnection(ConnectionStub(), TypeInfoStub())
+ cursor = zc.cursor()
+ cursor.execute('select * from blah')
+
+ self.assertEqual(zc._txn_registered, True)
+ self.assertEqual(len(t._resources), 1)
+
+ def test_commit(self):
+ t = get_transaction()
+ t.begin()
+ zc = ZopeConnection(ConnectionStub(), TypeInfoStub())
+ self._txn_registered = True
+ zc.commit()
+ self.assertEqual(zc._txn_registered, False,
+ "did not forget the transaction")
+
+ def test_rollback(self):
+ t = get_transaction()
+ t.begin()
+ zc = ZopeConnection(ConnectionStub(), TypeInfoStub())
+ self._txn_registered = True
+ zc.rollback()
+ self.assertEqual(zc._txn_registered, False,
+ "did not forget the transaction")
+
+ def test_getattr(self):
+ zc = ZopeConnection(ConnectionStub(), TypeInfoStub())
+ cursor = zc.cursor()
+
+ self.assertEqual(zc.answer(), 42)
+
+ def tearDown(self):
+ "Abort the transaction"
+ get_transaction().abort()
+
+def test_suite():
+ return makeSuite(ZopeConnectionTests)
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_zopecursor.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_zopecursor.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,115 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+
+from unittest import TestCase, TestSuite, main, makeSuite
+from zope.app.rdb import ZopeConnection
+from zope.app.interfaces.rdb import IZopeCursor
+from zope.app.rdb import ZopeCursor
+from zope.app.rdb.tests.stubs import *
+
+class MyConnectionStub(ConnectionStub):
+ def cursor(self):
+ return MyCursorStub()
+
+
+raw = [['mano', 2, 'buvo batai'],
+ ['dingo', 1, 'nerandu'],
+ ['as su', 1, 'batuku'],
+ ['eiti i', None, 'galiu']]
+
+converted = [['my', 42, 'shoes were'],
+ ['were lost', 41, "can't find"],
+ ['with', 41, 'shoe'],
+ ['go to', None, 'I can']]
+
+
+class MyCursorStub(CursorStub):
+
+ description = ((None, 'string'), (None, 'int'), (None, 'foo'))
+
+ def fetchone(self):
+ return raw[:1]
+
+ def fetchall(self):
+ return raw
+
+ def fetchmany(self, size=2):
+ return raw[:size]
+
+
+class MyTypeInfoStub(TypeInfoStub):
+
+ def getConverter(self, type):
+
+ def stringConverter(x):
+ return {'mano': 'my',
+ 'dingo': 'were lost',
+ 'as su': 'with',
+ 'eiti i': 'go to'}[x]
+
+ def intConverter(x):
+ if x is None:
+ return None
+ else:
+ return x + 40
+
+ def fooConverter(x):
+ return {'buvo batai': 'shoes were',
+ 'nerandu': "can't find",
+ 'batuku': 'shoe',
+ 'galiu': 'I can'}[x]
+
+ return {'string': stringConverter,
+ 'int': intConverter,
+ 'foo': fooConverter}[type]
+
+
+class ZopeCursorTests(TestCase):
+
+ def setUp(self):
+ zc = ZopeConnection(MyConnectionStub(), MyTypeInfoStub())
+ self.cursor = ZopeCursor(zc.conn.cursor(), zc)
+
+ def test_cursor_fetchone(self):
+ results = self.cursor.fetchone()
+ expected = converted[:1]
+ self.assertEqual(results, expected,
+ 'type conversion was not performed in cursor.fetchone:\n'
+ 'got %r, expected %r' % (results, expected))
+
+ def test_cursor_fetchmany(self):
+ results = self.cursor.fetchmany()
+ expected = converted[:2]
+ self.assertEqual(results, expected,
+ 'type conversion was not performed in cursor.fetchmany:\n'
+ 'got %r,\n'
+ 'expected %r' % (results, expected))
+
+ def test_cursor_fetchall(self):
+ results = self.cursor.fetchall()
+ expected = converted
+ self.assertEqual(results, expected,
+ 'type conversion was not performed in cursor.fetchall:\n'
+ 'got %r,\n'
+ 'expected %r' % (results, expected))
+
+
+def test_suite():
+ return makeSuite(ZopeCursorTests)
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
=== Zope3/src/zope/app/rdb/tests/test_zopedatabaseadapter.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:45 2002
+++ Zope3/src/zope/app/rdb/tests/test_zopedatabaseadapter.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,79 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+import unittest
+from zope.app.rdb import ZopeDatabaseAdapter
+from zope.app.rdb import ZopeConnection
+
+class ConnectionStub:
+
+ def close(self):
+ pass
+
+
+class DAStub(ZopeDatabaseAdapter):
+
+ def _connection_factory(self):
+ return ConnectionStub()
+
+
+class TestZopeDatabaseAdapter(unittest.TestCase):
+
+ def setUp(self):
+ self._da = DAStub('dbi://test')
+
+ def testSetGetDSN(self):
+ da = self._da
+ da.setDSN('dbi://foo')
+ self.assertEqual('dbi://foo', da.dsn)
+ self.assertEqual('dbi://foo', da.getDSN())
+
+ def testConnect(self):
+ da = self._da
+ da.connect()
+ self.assertEqual(ZopeConnection, da._v_connection.__class__)
+
+ def testDisconnect(self):
+ da = self._da
+ da.disconnect()
+ self.assertEqual(None, da._v_connection)
+
+ def testIsConnected(self):
+ da = self._da
+ da.connect()
+ self.assertEqual(1, da.isConnected())
+ da.disconnect()
+ self.assertEqual(0, da.isConnected())
+
+ def testCall(self):
+ da = self._da
+ conn = da()
+ self.assertEqual(ZopeConnection, conn.__class__)
+
+ def testGetConverter(self):
+ from zope.app.rdb import identity
+ da = self._da
+ conv = da.getConverter('any')
+ self.assert_(conv is identity, "default converter is wrong")
+
+
+def test_suite():
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.makeSuite(TestZopeDatabaseAdapter))
+ return suite
+
+if __name__ == '__main__':
+ unittest.TextTestRunner().run(test_suite())
=== Zope3/src/zope/app/rdb/tests/test_zopedbtransactionmanager.py 1.1 => 1.2 ===
--- /dev/null Wed Dec 25 09:13:46 2002
+++ Zope3/src/zope/app/rdb/tests/test_zopedbtransactionmanager.py Wed Dec 25 09:13:14 2002
@@ -0,0 +1,50 @@
+##############################################################################
+#
+# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+$Id$
+"""
+
+from unittest import TestCase, TestSuite, main, makeSuite
+from transaction import get_transaction
+from transaction.tests.abstestIDataManager import IDataManagerTests
+from zope.app.rdb import ZopeDBTransactionManager
+from zope.app.rdb import ZopeConnection
+from zope.app.rdb.tests.stubs import ConnectionStub, CursorStub, TypeInfoStub
+
+class TxnMgrTest(IDataManagerTests, TestCase):
+
+ def setUp(self):
+ self.conn = ConnectionStub()
+ zc = ZopeConnection(self.conn, TypeInfoStub())
+ self.datamgr = ZopeDBTransactionManager(zc)
+ zc.registerForTxn()
+ self.txn_factory = get_transaction
+
+ def tearDown(self):
+ """ make sure the global env is clean"""
+ get_transaction().abort()
+
+ def test_abort(self):
+ get_transaction().abort()
+ self.assertEqual(self.conn._called.get('rollback'), 1)
+
+ def test_commit(self):
+ get_transaction().commit()
+ self.assertEqual(self.conn._called.get('commit'), 1)
+
+def test_suite():
+ return makeSuite(TxnMgrTest)
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')