#! /usr/bin/env python
# -*- coding: iso-8859-1 -*-
#-----------------------------------------------------------------------------
# Modeling Framework: an Object-Relational Bridge for python
#
# Copyright (c) 2001-2004 Sbastien Bigaret <sbigaret@users.sourceforge.net>
# All rights reserved.
#
# This file is part of the Modeling Framework.
#
# This code is distributed under a "3-clause BSD"-style license;
# see the LICENSE file for details.
#-----------------------------------------------------------------------------
"""
Global tests for the Adaptor Layer
CVS information
$Id: test_AdaptorLayer.py 932 2004-07-20 06:21:57Z sbigaret $
"""
__version__='$Revision: 932 $'[11:-2]
import unittest, sys
import utils
if __name__ == "__main__":
utils.fixpath()
from Modeling import ModelSet,Model
from Modeling.EditingContext import EditingContext
from Modeling.FetchSpecification import FetchSpecification
from Modeling.Qualifier import qualifierWithQualifierFormat
from Modeling import Adaptor
from Modeling import Database
from testPackages.AuthorBooks.Writer import Writer
from testPackages.AuthorBooks.Book import Book
# utilities
def concreteAdaptorChannels_are_closed(anEntityName, ec):
"Returns 1 if they are all closed, 0 otherwise"
fs=FetchSpecification(anEntityName)
dbContext=ec.rootObjectStore().objectStoreForFetchSpecification(fs)
return not dbContext.adaptorContext().hasOpenChannels()
#for dbChannel in dbContext.registeredChannels():
# if dbChannel.adaptorChannel().isOpen():
# return 0
#return 1
import os
if not os.environ.get('MDL_TRANSIENT_DB_CONNECTION'):
os.environ['MDL_TRANSIENT_DB_CONNECTION']='True'
class Tests_AdaptorLayer_Global(unittest.TestCase):
"Global tests for the Adaptor Layer"
def test_01_adaptorChannels_are_closed(self):
"[AdaptorLayer] adaptorChannels should be closed"
# SELECT
ec=EditingContext()
fetchSpec=FetchSpecification(entityName='Writer')
objects=ec.objectsWithFetchSpecification(fetchSpec)
self.failUnless(concreteAdaptorChannels_are_closed('Writer', ec),
'Failed for SELECT')
# INSERT
w1=Writer()
w2=Writer()
w1.setLastName('test_insert1')
w2.setLastName('test_insert2')
ec.insertObject(w1); ec.insertObject(w2)
ec.saveChanges()
self.failUnless(concreteAdaptorChannels_are_closed('Writer', ec),
'Failed for INSERT')
# UPDATE
w1.setFirstName('test01')
ec.saveChanges()
self.failUnless(concreteAdaptorChannels_are_closed('Writer', ec),
'Failed for UPDATE')
# DELETE
ec.deleteObject(w2)
ec.saveChanges()
self.failUnless(concreteAdaptorChannels_are_closed('Writer', ec),
'Failed for DELETE')
def tearDown(self):
"""
Cleans up the Database after each tests
"""
ec=EditingContext()
w=Writer()
dbChannel=ec.rootObjectStore().objectStoreForObject(w).availableChannel()
channel=dbChannel.adaptorChannel()
sqlExpr=channel.adaptorContext().adaptor().expressionClass()()
sqlExpr.setStatement("delete from BOOK where id>4")
channel.evaluateExpression(sqlExpr)
sqlExpr.setStatement("delete from WRITER where id>3")
channel.evaluateExpression(sqlExpr)
def usage(prgName):
_usage="""%s [-vV] [-p] [-d <dbAdaptorName>]
Runs the tests for the Modeling package
Options
--------
-v Minimally verbose
-V Really verbose
-h Prints this message
-p enables profiling
-d sets the database adaptor to use: Postgresql (default), MySQL or SQLite
IMPORTANT: these tests requires that test_EditingContext_Global.py was
previously executed with the '-r' option.
""" % prgName
sys.stderr.write(_usage)
verbose=0
database_cfg='Postgresql.cfg'
def main(args):
me=args[0]
import getopt
options, args = getopt.getopt(sys.argv[1:], 'vVpd:')
global verbose, database_cfg
profile=0
for k, v in options:
if k=='-h': usage(me); return 1
if k=='-v': verbose=1; continue
if k=='-V': verbose="Y"; continue
if k=='-p': profile=1; continue
if k=='-d':
if v not in ('Postgresql', 'MySQL', 'SQLite'): usage(me); return 1
database_cfg='%s.cfg'%v
continue
if args: usage(me); return 1
author_books_model=ModelSet.defaultModelSet().modelNamed('AuthorBooks')
Model.updateModelWithCFG(author_books_model, database_cfg)
# MySQL specifics: change TIMESTAMP to DATETIME
if database_cfg=='MySQL.cfg':
author_books_model.entityNamed('Writer').attributeNamed('birthday').setExternalType('DATETIME')
if profile:
import profile
profile.run('utils.run_suite(test_suite(), verbosity=verbose)',
'profile.out')
return
else:
return utils.run_suite(test_suite(), verbosity=verbose)
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(Tests_AdaptorLayer_Global, "test_"))
return suite
if __name__ == "__main__":
errs = main(sys.argv)
sys.exit(errs and 1 or 0)
|