[Zope3-checkins] CVS: Zope3/src/zope/app/rdb/tests - __init__.py:1.1.2.1 stubs.py:1.1.2.1 test_dsnparser.py:1.1.2.1 test_gadflyadapter.py:1.1.2.1 test_resultset.py:1.1.2.1 test_row.py:1.1.2.1 test_sqlcommand.py:1.1.2.1 test_zopeconnection.py:1.1.2.1 test_zopecursor.py:1.1.2.1 test_zopedatabaseadapter.py:1.1.2.1 test_zopedbtransactionmanager.py:1.1.2.1
Jim Fulton
jim@zope.com
Mon, 23 Dec 2002 14:32:15 -0500
Update of /cvs-repository/Zope3/src/zope/app/rdb/tests
In directory cvs.zope.org:/tmp/cvs-serv19908/zope/app/rdb/tests
Added Files:
Tag: NameGeddon-branch
__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:
Initial renaming before debugging
=== Added File Zope3/src/zope/app/rdb/tests/__init__.py ===
#
# This file is necessary to make this directory a package.
=== Added File Zope3/src/zope/app/rdb/tests/stubs.py ===
##############################################################################
#
# 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: stubs.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
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
=== Added File Zope3/src/zope/app/rdb/tests/test_dsnparser.py ===
##############################################################################
#
# 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: test_dsnparser.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
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())
=== Added File Zope3/src/zope/app/rdb/tests/test_gadflyadapter.py ===
##############################################################################
#
# 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: test_gadflyadapter.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
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')
=== Added File Zope3/src/zope/app/rdb/tests/test_resultset.py ===
##############################################################################
#
# 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: test_resultset.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
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')
=== Added File Zope3/src/zope/app/rdb/tests/test_row.py ===
##############################################################################
#
# 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: test_row.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
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')
=== Added File Zope3/src/zope/app/rdb/tests/test_sqlcommand.py ===
##############################################################################
#
# 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: test_sqlcommand.py,v 1.1.2.1 2002/12/23 19:32:12 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.rdb import SQLCommand
from zope.app.interfaces.rdb import IConnectionService
from zope.app.interfaces.rdb import IZopeConnection
from zope.app.interfaces.rdb import IZopeCursor
from Zope.App.ComponentArchitecture import NextService
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.component.service import \
serviceManager as sm
# Make spme 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 Test(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 testSimpleSQLCommand(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 makeSuite(Test)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/rdb/tests/test_zopeconnection.py ===
##############################################################################
#
# 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: test_zopeconnection.py,v 1.1.2.1 2002/12/23 19:32:13 jim Exp $
"""
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')
=== Added File Zope3/src/zope/app/rdb/tests/test_zopecursor.py ===
##############################################################################
#
# 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: test_zopecursor.py,v 1.1.2.1 2002/12/23 19:32:13 jim Exp $
"""
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')
=== Added File Zope3/src/zope/app/rdb/tests/test_zopedatabaseadapter.py ===
##############################################################################
#
# 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: test_zopedatabaseadapter.py,v 1.1.2.1 2002/12/23 19:32:13 jim Exp $
"""
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())
=== Added File Zope3/src/zope/app/rdb/tests/test_zopedbtransactionmanager.py ===
##############################################################################
#
# 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: test_zopedbtransactionmanager.py,v 1.1.2.1 2002/12/23 19:32:13 jim Exp $
"""
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')