[Zope-Checkins] CVS: Zope3/lib/python/Zope/App/RDB/tests - Stubs.py:1.1 __init__.py:1.1 testRow.py:1.1 testZopeConnection.py:1.1 testZopeDBTransactionManager.py:1.1

Kapil Thangavelu kvthan@wm.edu
Tue, 25 Jun 2002 11:41:46 -0400


Update of /cvs-repository/Zope3/lib/python/Zope/App/RDB/tests
In directory cvs.zope.org:/tmp/cvs-serv5574/tests

Added Files:
	Stubs.py __init__.py testRow.py testZopeConnection.py 
	testZopeDBTransactionManager.py 
Log Message:
lots of work on rdb :-) almost there.




=== Added File Zope3/lib/python/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 2002/06/25 15:41:45 k_vertigo 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


    


=== Added File Zope3/lib/python/Zope/App/RDB/tests/__init__.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.
# 
##############################################################################


=== Added File Zope3/lib/python/Zope/App/RDB/tests/testRow.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.
#
##############################################################################
"""XXX short summary goes here.

XXX longer description goes here.

$Id: testRow.py,v 1.1 2002/06/25 15:41:46 k_vertigo Exp $
"""

from Zope.App.RDB.Row import row_class_factory
from Zope.Security.Proxy import ProxyFactory
from Zope.Exceptions import ForbiddenAttribute

from unittest import TestCase, TestSuite, main, makeSuite

class RowTests(TestCase):

    def test_row_klass_creation(self):

        columns = ('food', 'name')
        data = ('pizza', 'john')
        
        klass = row_class_factory(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_row_klass_security_declarations(self):

        columns = ('type', 'speed')
        data = ('airplane', '800km')
        
        klass = row_class_factory(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_suite():
    return TestSuite((
        makeSuite(RowTests),
        ))

if __name__=='__main__':
    main(defaultTest='test_suite')


=== Added File Zope3/lib/python/Zope/App/RDB/tests/testZopeConnection.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.
#
##############################################################################
"""XXX short summary goes here.

XXX longer description goes here.

$Id: testZopeConnection.py,v 1.1 2002/06/25 15:41:46 k_vertigo Exp $
"""

from unittest import TestCase, TestSuite, main, makeSuite
from Transaction import get_transaction
from Zope.App.RDB.ZopeConnection import ZopeConnection
from Zope.App.RDB.IZopeCursor import IZopeCursor
from Stubs import *

class ZopeConnectionTests(TestCase):

    def test_cursor(self):
        "ZopeConnection.cursor() should return an IZopeCursor"
        zc = ZopeConnection(ConnectionStub())
        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())
        cursor = zc.cursor()
        cursor.execute('select * from blah')
        
        self.failUnless(zc._txn_registered == 1,
                        """connection was not registered for txn (conn)""")

        self.failUnless(len(t._objects) == 1,
                        """connection was not registered for txn (txn) """)

    def test_getattr(self):
        "ZopeConnection must reveal Connection's methods"

        zc = ZopeConnection(ConnectionStub())
        cursor = zc.cursor()

        self.failUnless(zc.answer() == 42, "Cannot see the connection")


    def tearDown(self):
        "Abort the transaction"
        get_transaction().abort()
        
def test_suite():
    return TestSuite((
        makeSuite(ZopeConnectionTests),
        ))

if __name__=='__main__':
    main(defaultTest='test_suite')






=== Added File Zope3/lib/python/Zope/App/RDB/tests/testZopeDBTransactionManager.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.
#
##############################################################################
"""XXX short summary goes here.

XXX longer description goes here.

$Id: testZopeDBTransactionManager.py,v 1.1 2002/06/25 15:41:46 k_vertigo Exp $
"""

from unittest import TestCase, TestSuite, main, makeSuite
from Transaction import get_transaction
from Zope.App.RDB.ZopeDBTransactionManager import ZopeDBTransactionManager
from Zope.App.RDB.ZopeConnection import ZopeConnection
from Stubs import *

class TxnMgrTest(TestCase):
    """
    test txn integration.
    """

    def tearDown(self):
        """ make sure the global env is clean"""
        get_transaction().abort()

    def test_abort(self):
	
        conn = ConnectionStub()
        zc = ZopeConnection(conn)
        tm = ZopeDBTransactionManager(zc)

        zc.registerForTxn()
        
        t = get_transaction()
        t.abort()

        self.failUnless(conn._called.get('rollback')==1,
                        """ abort failed """)

    def test_commit(self):
        
        conn = ConnectionStub()
        zc = ZopeConnection(conn)
        tm = ZopeDBTransactionManager(zc)

        zc.registerForTxn()
        
        t = get_transaction()
        t.commit()

        self.failUnless(conn._called.get('commit')==1,
                        """ commit failed """)

def test_suite():
    return TestSuite((
        makeSuite(TxnMgrTest),
        ))

if __name__=='__main__':
    main(defaultTest='test_suite')