[Zope-Checkins] CVS: Products/DCOracle2/test - test_zoracleda_basics.py:1.1 test_zoracleda.py:1.1 test_zopeda.py:1.1

[Chris Withers]

Update of /cvs-repository/Products/DCOracle2/test
In directory cvs.zope.org:/tmp/cvs-serv12067

Added Files:
	test_zoracleda_basics.py test_zoracleda.py test_zopeda.py 
Log Message:
New ZOracleDA test suite.


=== Added File Products/DCOracle2/test/test_zoracleda_basics.py ===
import common
import test_zopeda
import unittest

class test_ZOracleDA(test_zopeda.DATests):

    def _createDA(self,folder,id):
        
        from Products.ZOracleDA.DA import manage_addZOracleConnection

        manage_addZOracleConnection(
            folder,
            id,
            '',
            common.getConnectionString()
            )

    def _getConnectionString(self):
        return common.getConnectionString()

    create_sql = (
        "drop table users",
        "create table users (name varchar(10), password varchar(10), role varchar(10))",
        )
    

if __name__ == '__main__':
    unittest.main()


=== Added File Products/DCOracle2/test/test_zoracleda.py ===
import unittest

def test_suite():

    import test_dcoracle2
    import test_zoracleda_basics
    
    return unittest.TestSuite((
        test_dcoracle2.test_suite(),
        unittest.makeSuite(test_zoracleda_basics.test_ZOracleDA),
        ))

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


=== Added File Products/DCOracle2/test/test_zopeda.py ===
# This file defines the DATests class.

# This is an abstract base for testing a
# Zope Database Adapter's basic functionality

import unittest
import Zope

from Testing.makerequest import makerequest
from types import StringType

class DATests(unittest.TestCase):

    def _createDA(self,folder,id):
        # this should create a Database Adapter object
        # with the id supplied of the appropriate type
        # in the folder object supplied 
        raise NotImplementedError

    def _getConnectionString(self):
        # this should return the connection string
        # for use during testing
        raise NotImplementedError
    
    # the following SQL should be overridden in the subclass
    # if the database under test requires difference SQL
    create_sql = (
        'drop table user',
        'create table users (name varchar, password varchar, role varchar)',
        )

    insert_sql = (
        "insert into users(name,password,role) values ('fred','password','God')",
        "insert into users(name,password,role) values ('john','monkey','Manager')",
        "insert into users(name,password,role) values ('chris','wibble','Enslaved')",
        )

    select_sql = (
        "select * from users",
        )
    
    # The stuff below should only need to be changed when subclassing
    # if the database adapter under test does something non-standard.
    
    folder_id = 'test'
    
    def setUp( self ):
        if hasattr(Zope, 'startup'):
            Zope.startup()
        get_transaction().begin()
        self.connection = Zope.DB.open()
        root = self.root = makerequest(self.connection.root()[ 'Application' ])
        self.REQUEST  = root.REQUEST
        self.RESPONSE = root.REQUEST.RESPONSE
        try:
            self.root._delObject(self.folder_id)
        except AttributeError:
            pass
        self.root.manage_addFolder(self.folder_id)
        self.folder = self.root[self.folder_id]
        self._createDA(self.folder,'test_connection')
        self.da = self.folder.test_connection

    def tearDown( self ):
        get_transaction().abort()
        self.connection.close()

    ### UI forms
    def test_manage_main(self):
        self.da.manage_main(None,self.REQUEST)

    def test_manage_properties(self):
        self.da.manage_properties(None,self.REQUEST)

    def test_manage_testForm(self):
        self.da.manage_testForm(None,self.REQUEST)

    ### support methods

    def test_title_and_id(self):
        self.failUnless(
            isinstance(self.da.title_and_id(),StringType)
                       )
        
    def test_title_or_id(self):
        self.failUnless(
            isinstance(self.da.title_and_id(),StringType)
                       )
        
    def test_connected(self):
        self.failIf(self.da.connected())
        self.da()
        self.failUnless(self.da.connected())
        
    def test_edit(self):
        self.da.edit('New Title',
                     self._getConnectionString(),
                     1)
        self.assertEqual(self.da.title,'New Title')
        # can't really check whether connection_String
        # and check paramters have any effect :-S

    def test_manage_edit(self):
        r = self.da.manage_edit('New Title',
                                self._getConnectionString(),
                                1,
                                self.REQUEST)
        self.failUnless(r,StringType)
        self.assertEqual(self.da.title,'New Title')
        # can't really check whether connection_String
        # and check paramters have any effect :-S

    def test_close_connection(self):
        self.da()
        self.failUnless(self.da.connected())
        self.da.manage_close_connection(self.REQUEST)
        self.failIf(self.da.connected())

    def test_open_connection(self):
        self.failIf(self.da.connected())
        self.da.manage_open_connection()
        self.failUnless(self.da.connected())

    def test_connect(self):
        self.failIf(self.da.connected())
        self.da.connect(self._getConnectionString())
        self.failUnless(self.da.connected())

    def test_sql_quote(self):
        self.assertEqual(
            self.da.sql_quote__("'x'"),
            "'''x'''"
            )
        
    ### functional stuff such as working with ZSQL methods
        
    def test_create(self):
        # creation is done in setUp
        pass
    
    def test_SQLConnectionIDs(self):
        from Acquisition import aq_base
        from Products.ZSQLMethods.SQL import SQLConnectionIDs
        self.failUnless(
            ('test_connection','test_connection') in SQLConnectionIDs(self.folder)
            )
        # paranoid check to see if 'test_conection' is really what
        # we're after. Since we've created the test_connection in
        # self.folder, all we really need to do is make sure it's not being
        # acquired.
        self.failUnless(hasattr(aq_base(self.folder),'test_connection'))

    def test_call(self):
        from Acquisition import aq_base
        # call the DA, should return a DB connection surrogate
        self.dbc = aq_base(self.da())
        # does it have a query method?
        self.failUnless(hasattr(self.dbc,'query'))
        # do it have TM methods?

        ### FIXME: How do we actually check that these work?!
        for name in [
            '_begin',
            '_register',
            'tpc_begin',
            'commit',
            '_finish',
            '_abort',
            'tpc_vote',
            'tpc_finish',
            'abort',
            'tpc_abort',
            'sortKey',
            ]:
            self.failUnless(hasattr(self.dbc,name),name)

    def test_manage_test(self):
        self.test_insertTable()
        # this also tests the result building bits
        from Shared.DC.ZRDB.Results import Results
        r = self.da.manage_test(self.select_sql[0])
        self.failUnless(isinstance(r,Results))
        self.assertEqual(len(r),3)
        self.assertEqual(
            r.dictionaries(),
            [{'NAME': 'fred', 'ROLE': 'God', 'PASSWORD': 'password'},
             {'NAME': 'john', 'ROLE': 'Manager', 'PASSWORD': 'monkey'},
             {'NAME': 'chris', 'ROLE': 'Enslaved', 'PASSWORD': 'wibble'}]
            )
        
    def _callSQL(self,attributename):
        results = []
        for statement in getattr(self,attributename):
            results.append(self.da().query(statement))
        return results
    
    def test_createTable(self):
        self.assertEqual(
            self._callSQL('create_sql'),
            [((), ()), ((), ())],
            )
    
    def test_insertTable(self):
        self.test_createTable()
        self.assertEqual(
            self._callSQL('insert_sql'),
            [((), ()), ((), ()), ((), ())],
            )
    
    def test_selectTable(self):
        self.test_insertTable()
        self.assertEqual(
            self._callSQL('select_sql'),
            [
            ([{'name': 'NAME', 'type': 's', 'null': 1, 'width': 10},
              {'name': 'PASSWORD', 'type': 's', 'null': 1, 'width': 10},
              {'name': 'ROLE', 'type': 's', 'null': 1, 'width': 10}],
             [['fred', 'password', 'God'],
              ['john', 'monkey', 'Manager'],
              ['chris', 'wibble', 'Enslaved']])
            ]
            )

    def test_selectTableMaxRows(self):
        self.test_insertTable()
        # second parameter to query method is max rows
        self.assertEqual(
            self.da().query(self.select_sql[0],2),
            ([{'name': 'NAME', 'type': 's', 'null': 1, 'width': 10},
              {'name': 'PASSWORD', 'type': 's', 'null': 1, 'width': 10},
              {'name': 'ROLE', 'type': 's', 'null': 1, 'width': 10}],
             [['fred', 'password', 'God'],
              ['john', 'monkey', 'Manager']])
            )