Happy New Year, readers! The bulk of today's code is shared at dropbox.com/u/1917253/site-packages/MySQLConnector.py.
So the last leg of our journey into Python-database connectivity is (finally)
the MySQLConnector class. There's been a lot of what I'd
call "support" development to get to this point:
- Three interfaces (
IsDataConnector,IsQuery, andIsResultSet); - One abstract class (
BaseDataConnector); and - Three concrete classes (
Query,ResultSetandRecord).
DataConnectors module is just barely closer to 2,400 lines
of code than 2,300, in about a 40:60 split between "real" code and test-code.
One (hopefully significant) advantage of this approach is that the creation
of the MySQLConnector class is going to be crazy-simple. As a
derivative of both BaseDataConnector and IsConfigurable,
there's really not a whole lot of code that's specific to the MySQLConnector
to be written, and other similar classes (say, a hypothetical PostGreSQLConnector)
would require a similarly small level of effort. Consider that MySQLConnector
has the following properties and methods:
- The
Configuremethod, fromIsConfigurable, which may need specific implementation, but that was implemented at a basic level inBaseDataConnector. - The
Connectmethod, required byIsDataConnector, which will need implementation. - The
Databaseproperty, fromBaseDataConnector, that is complete and tested until or unless we need to make implementation-specific changes. - The
Executemethod, required byIsDataConnector, which will need implementation. - The
Hostproperty, fromBaseDataConnector, also complete and tested, barring implementation-specific changes. - The
Passwordproperty, fromBaseDataConnector, also complete and tested, barring implementation-specific changes. - The
Userproperty, fromBaseDataConnector, also complete and tested, barring implementation-specific changes. - And, of course, unit-tests for the new class. Ideally, there should also be some system tests to test/prove out functionality against an actual database.
As I was working on MySQLConnector, I discovered that I'd never
finished figuring out how to unit-test the Configure method of it,
and that I was missing a Configuration class:
Configuration (from the Configuration module):
class Configuration( ConfigParser.ConfigParser, object ):
"""Represents configuration data derived from a text-file."""
#################################
# Class Attributes #
#################################
#################################
# Class Property-Getter Methods #
#################################
def _GetConfigFile( self ):
"""Gets the config-file-path specified at object creation."""
return self._configFile
def _GetSections( self ):
"""Gets the available configuration-sections."""
if self._configSections == None:
try:
self.read( self._configFile )
except Exception, error:
raise AttributeError( '%s.Sections: Could not read configuration from %s. Original error:\n\t%s' % ( self.__class__.__name__, self._configFile, error ) )
self._configSections = {}
for theSection in self.sections():
self._configSections[ theSection ] = {}
for theItem in self.items( theSection, 1 ):
itemName = theItem[0]
itemValue = theItem[1]
if itemValue[0] in [ '[', '{', '(' ]:
# Evaluate the value instead... Ugly, but viable...
itemValue = eval( itemValue )
self._configSections[ theSection ][ itemName ] = itemValue
return self._configSections
#################################
# Class Property-Setter Methods #
#################################
#################################
# Class Properties #
#################################
ConfigFile = property( _GetConfigFile, None, None, _GetConfigFile.__doc__ )
Sections = property( _GetSections, None, None, _GetSections.__doc__ )
#################################
# Object Constructor #
#################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'filePath', None, '(String) The path to the configuration-file.' )
def __init__( self, filePath ):
"""Object constructor."""
ConfigParser.ConfigParser.__init__( self )
self._configSections = None
self._configFile = filePath
#################################
# Object Destructor #
#################################
#################################
# Class Methods #
#################################
__all__ += [ 'Configuration' ]
This will likely need to be refactored at some point, since there are other
configuration mechanisms than the ConfigParser base here. For now, though,
I'll leave it be. Configuration instances are basically just wrappers
around ConfigParser instances, that provide ConfigFile
and Sections properties. The ConfigFile is probably
pretty straightforward - it's a pointer to the configuration-file that the
instance uses. Sections might be a little harder to understand,
though - I wanted to have a simple access mechanism for the configuration-sections
of a given configuration-file, in order to be able to retrieve section-items
without having to make a function-call every time.
With that in place, I could finish the unit-tests for BaseDataConnector:
def testConfigure( self ):
"""Unit-tests the Configure method of the BaseDataConnector abstract class."""
# First, we need to build out a configuration-file
hostName = 'hostName'
databaseName = 'databaseName'
userName = 'userName'
userPassword = 'userPassword'
configString = """[Default Section]
SectionName1: SectionValue1
SectionName2: SectionValue2
[Data Access]
host: %s
database: %s
user: %s
password: %s
""" % ( hostName, databaseName, userName, userPassword )
fp = open( 'test.cfg', 'w' )
fp.write( configString )
fp.close()
# Now we need to create the configuration object and read in the configuration values
configParser = Configuration( 'test.cfg' )
# Now, finally, we can test the configuration
testObject = BaseDataConnectorDerived()
testObject.Configure( configParser, 'Data Access' )
self.assertEquals( testObject.Host, hostName, 'The host name should be retrievable from configuration' )
self.assertEquals( testObject.Database, databaseName, 'The database name should be retrievable from configuration' )
self.assertEquals( testObject.User, userName, 'The user name should be retrievable from configuration' )
self.assertEquals( testObject.Password, userPassword, 'The user password should be retrievable from configuration' )
os.unlink( 'test.cfg' )
All this is doing is creating a hard-coded (and very simple) configuration-file,
with a Data Access section that contains the host, database and user
names for the connector, and the password that it should use. Looking at lines 22 and
24-25, creation of a BaseDataConnector-derived instance should
always be feasible with this sort of code-structure: Create an instance with no
arguments, create a configuration instance, then call the Configure
method of the data-connector instance.
It may bear noting that MySQLConnector doesn't have any sort of
direct support for cursors - that may be something that I'll add in to
BaseDataConnector at some point in the future, if there's
support for it across MySQL (yes), PostgreSQL (not sure), and ODBC (also not sure),
which are the three data-connection types I expect this to support in the long
run.
MySQLConnector (Nominal final class):
#####################################
# Convenience imports #
#####################################
# See To-do list on MySQLConnector.__init__
#####################################
# Classes defined in the module #
#####################################
class MySQLConnector( BaseDataConnector, IsConfigurable, object ):
"""Class doc-string."""
##################################
# Class Attributes #
##################################
##################################
# Class Property-Getter Methods #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
def _GetConnection( self ):
"""Gets the connection to the specified MySQL database."""
if self._connection == None:
self.Connect()
return self._connection
##################################
# Class Property-Setter Methods #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'value', None, 'The name of the database the connection will be made to.' )
def _SetDatabase( self, value ):
if self._connection != None:
raise AttributeError( '%s.Database cannot be reset once a connection is established' % ( self.__class__.__name__ ) )
BaseDataConnector._SetDatabase( self, value )
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'value', None, 'The name or IP address of the host that the database being connected to lives on.' )
def _SetHost( self, value ):
if self._connection != None:
raise AttributeError( '%s.Host cannot be reset once a connection is established' % ( self.__class__.__name__ ) )
BaseDataConnector._SetHost( self, value )
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'value', None, 'The password that will be used to connect to the database.' )
def _SetPassword( self, value ):
if self._connection != None:
raise AttributeError( '%s.Password cannot be reset once a connection is established' % ( self.__class__.__name__ ) )
BaseDataConnector._SetPassword( self, value )
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'value', None, 'The user-name that will be used to connect to the database.' )
def _SetUser( self, value ):
if self._connection != None:
raise AttributeError( '%s.User cannot be reset once a connection is established' % ( self.__class__.__name__ ) )
BaseDataConnector._SetUser( self, value )
##################################
# Class Property-Deleter Methods #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
def _DelConnection( self ):
self._connection = None
##################################
# Class Properties #
##################################
Connection = property( _GetConnection, None, None, 'Gets the connection to the specified MySQL database.' )
Database = property( BaseDataConnector._GetDatabase, _SetDatabase, BaseDataConnector._DelDatabase, BaseDataConnector._GetDatabase.__doc__ )
Host = property( BaseDataConnector._GetHost, _SetHost, BaseDataConnector._DelHost, BaseDataConnector._GetHost.__doc__ )
Password = property( BaseDataConnector._GetPassword, _SetPassword, BaseDataConnector._DelPassword, BaseDataConnector._GetPassword.__doc__ )
User = property( BaseDataConnector._GetUser, _SetUser, BaseDataConnector._DelUser, BaseDataConnector._GetUser.__doc__ )
##################################
# Object Constructor #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'keyword', 'parameters', 'host', 'The name or IP-address of the host that the database resides upon.' )
@DocumentArgument( 'keyword', 'parameters', 'database', 'The name of the database that the connection will be made to.' )
@DocumentArgument( 'keyword', 'parameters', 'user', 'The user-name used to connect to the database.' )
@DocumentArgument( 'keyword', 'parameters', 'password', 'The password used to connect to the database.' )
@ToDo( 'Figure out what items from DataConnectors should be imported for convenience in using the MySQLConnector class.' )
def __init__( self, **parameters ):
"""Object constructor."""
# Nominally final: Don't allow any class other than this one
if self.__class__ != MySQLConnector:
raise NotImplementedError( 'MySQLConnector is (nominally) a final class, and is not intended to be derived from.' )
self._connection = None
BaseDataConnector.__init__( self, **parameters )
##################################
# Object Destructor #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
def __del__( self ):
self._Close()
##################################
# Class Methods #
##################################
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
def _Close( self ):
if self._connection:
self._connection.close()
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
def Connect( self ):
"""Connects to the database specified by Database, residing on Host, using User and Password to connect."""
self._connection = MySQLdb.connect(
host=self._host,
db=self._database,
user=self._user,
passwd=self._password
)
@DocumentArgument( 'argument', 'self', None, SelfDocumentationString )
@DocumentArgument( 'argument', 'query', None, 'The IsQuery object that will be executed against the object\'s Connection.' )
def Execute( self, query ):
"""Executes the provided IsQuery object's query against the object's connection, returning one or more results."""
if not isinstance( query, IsQuery ):
raise TypeError( '%s.Execute expects an instance of IsQuery as it\'s query argument.' % ( self.__class__.__name__ ) )
self.Connection.query( query.Sql )
resultSets = []
resultSet = []
results = self.Connection.store_result()
while results:
row = list( results.fetch_row( 1, 1 ) )
while row:
resultSet += row
row = results.fetch_row( 1, 1 )
if self.Connection.next_result() == 0:
resultSets.append( resultSet )
resultSet = []
results = self.Connection.store_result()
else:
break
resultSets.append( resultSet )
return resultSets
__all__ += [ 'MySQLConnector' ]
- Line(s)
- 5
- This is a fine-tuning item that will almost certainly evolve in an obvious
way as I start writing code that actually uses
MySQLConnector: There will almost certainly be some (small?) set of imports from the baseDataConnectorsmodule that will be needed frequently enough to warrant importing them as part of theMySQLConnectorimport. Off the top of my head, I'd expect at leastQueryand maybeResultSetto be needed, but I'll wait to see what actually crops up as I start using the module... - 23-27
- An almost-typical property getter method, the main difference being that
the underlying MySQLdb
Connectionobject isn't instantiated until it's needed. - 33-59
- Overridden property-setter methods. The only additional functionality that they provide is to prevent the modification of any connection-control property (the host, database- and user-name, and password used for the connection) once the connection has been instantiated. The rationale for this is, basically, that once a connection has been established, changes to that connection should not be allowed. Chances are good that an error of some type would be thrown anyway, but since I'd rather raise errors as close to their "real" source as possible, I'm explicitly causing that to happen here.
- 73-77
- The actual property declarations. Since most of these are a mix of the
functionality from
BaseDataConnector(for the_Get...methods) and local implementations (_Set...methods), there's a mix of complete external method-references and "normal" local references to accommodate. - 101-103, 109-112
- I'm not sure that a formal object-destructor's actually needed,
since the underlying MySQLdb
Connectionobject probably closes the database connection all on it's own, but to be safe, I'm providing one that calls to the_Closemethod to make sure that the connection is closed cleanly when the object is destroyed. - The
_Closemethod is necessary because when the destructor fires, it doesn't recognize object properties. I think I mentioned this same pattern somewhere before, but I cannot recall where, exactly. - 114-122
- A public method to open the database-connection with the instance's parameters.
- 124-144
- The method that actually queries against the database, returning a list of
lists of dictionaries, ready to be passed to
ResultSetconstructors.- 128-129
- Typical type-checking of the supplied
queryargument. - 130
- Run the provided Query's SQL against the database, readying the results for formatting/digestion.
- 131, 132
- Prep the lists of all results, and the first list of rows for use.
- 133
- Get (and store) the first result-set returned.
- This uses
store_resultinstead ofuse_result, in the belief that a well-designed application (and database) should rarely return so many records that storing them wouldn't provide faster throughput than keeping the database-connection open and retrieving result-sets one at a time. - 134
- For so long as there are results:
- 135
- Get the first row of the current result-set
- 136-137
- For so long as there's a valid row, add the row to the current result-set, and get the next row.
- 139-142
- Once the current result's row-set has been processed, if there is another result-set after the current one, append the current results to the list to be returned, reset the row-list, and get/store the next result-set.
- 143-144
- If there are no other result-sets, then exit the loop.
- 145
- Make sure that the last set of rows gets attached to the results to be returned before returning them.
Unit-tests
Some of the unit-tests (the ones that are very basic system-tests, all in the testExecute method) require some actual tables on the test_database database:
CREATE TABLE `test_database`.`test_table_1` (
`id` INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'The record-id',
`first_name` VARCHAR(30) NOT NULL COMMENT 'A first name',
`last_name` varchar(30) NOT NULL COMMENT 'A last-name',
PRIMARY KEY (`id`)
)
ENGINE = MyISAM
COMMENT = 'A test-table';
INSERT INTO test_table_1 ( first_name, last_name )
VALUES ( 'Brian', 'Allbee' ), ( 'Joe', 'Smith' );
CREATE TABLE `test_database`.`test_table_2` (
`id` int UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'Record-ID',
`name_id` int NOT NULL
COMMENT 'ID of the record from test_table_1 that relates to this record',
`email` varchar(60) NOT NULL COMMENT 'An email address',
PRIMARY KEY (`id`)
)
ENGINE = MyISAM
COMMENT = 'Another test table';
INSERT INTO test_table_2 ( name_id, email )
VALUES ( 1, 'brian.allbee@somedomain.com' ), ( 2, 'joe.smith@somedomain.com' );
class MySQLConnectorDerived( MySQLConnector ):
def __init__( self, **parameters ):
MySQLConnector.__init__( self, **parameters )
def __del__( self ):
try:
MySQLConnector.__del__( self )
except:
pass
class testMySQLConnector( unittest.TestCase ):
"""Unit-tests the MySQLConnector class."""
def setUp( self ):
pass
def tearDown( self ):
pass
def getLiveConnection( self ):
"""Returns a MySQLConnector instance pointing to a common local test-database."""
return MySQLConnector( host='localhost', database='test_database', user='test_user', password='test_password' )
def testFinal( self ):
"""Testing final nature of the MySQLConnector class."""
try:
testObject = MySQLConnectorDerived()
self.fail( 'MySQLConnector is nominally a final class, and should not be extendable.' )
except NotImplementedError:
pass
except Exception, error:
self.fail( 'Instantiating a derivation of MySQLConnection should raise NotImplementedError, but %s was raised instead:\n %s' % ( error.__class__.__name__, error ) )
def testConstruction( self ):
"""Testing construction of the MySQLConnector class."""
testObject = MySQLConnector()
self.assertTrue( isinstance( testObject, MySQLConnector ), 'Instances of MySQLConnector should be instances of MySQLConnector.' )
self.assertEquals( testObject.Host, None, 'With no host supplied at construction, it\'s property value should be None' )
self.assertEquals( testObject.Database, None, 'With no database supplied at construction, it\'s property value should be None' )
self.assertEquals( testObject.User, None, 'With no user supplied at construction, it\'s property value should be None' )
self.assertEquals( testObject.Password, None, 'With no password supplied at construction, it\'s property value should be None' )
testHost = 'testHost'
testDatabase = 'database'
testUser = 'testUser'
testPassword = 'testPassword'
testObject = MySQLConnector( host=testHost, database=testDatabase, user=testUser, password=testPassword )
self.assertEquals( testObject.Host, testHost, 'The host supplied at construction should be present in the object\'s properties' )
self.assertEquals( testObject.Database, testDatabase, 'The database supplied at construction should be present in the object\'s properties' )
self.assertEquals( testObject.User, testUser, 'The user supplied at construction should be present in the object\'s properties' )
self.assertEquals( testObject.Password, testPassword, 'The password supplied at construction should be present in the object\'s properties' )
def testPropertyCountAndTests( self ):
"""Testing the properties of the MySQLConnector class."""
items = getMemberNames( MySQLConnector )[0]
actual = len( items )
expected = 5
self.assertEquals( expected, actual, 'MySQLConnector is expected to have %d properties to test, but %d were dicovered by inspection.' % ( expected, actual ) )
for item in items:
self.assertTrue( HasTestFor( self, item ), 'There should be a test for the %s property (test%s), but none was identifiable.' % ( item, item ) )
def testMethodCountAndTests( self ):
"""Testing the methods of the MySQLConnector class."""
items = getMemberNames( MySQLConnector )[1]
actual = len( items )
expected = 3
self.assertEquals( expected, actual, 'MySQLConnector is expected to have %d methods to test, but %d were dicovered by inspection.' % ( expected, actual ) )
for item in items:
self.assertTrue( HasTestFor( self, item ), 'There should be a test for the %s method (test%s), but none was identifiable.' % ( item, item ) )
# Unit-test properties
def testConnection( self ):
"""Unit-tests the Connection property of the MySQLConnector class."""
testObject = MySQLConnector( host='localhost', database='no_such_database', user='test_user', password='test_password' )
try:
testObject.Connect()
self.fail( 'Trying to connect to a non-existant database should raise OperationalError.' )
except MySQLdb.OperationalError:
pass
except Exception, error:
self.fail( 'Trying to connect to a non-existant database should raise OperationalError, but %s was raised instead:\n %s.' % ( error.__class__.__name__, error ) )
def testDatabase( self ):
"""Unit-tests the Database property of the MySQLConnector class."""
testObject = self.getLiveConnection()
oldDatabase = testObject.Database
testObject.Database = 'ook'
self.assertEquals( testObject.Database, 'ook', 'Before a connection is made, the object should allow modification of the Database property.' )
testObject.Database = oldDatabase
testObject.Connect()
try:
testObject.Database = 'ook'
self.fail( 'Once a connection is established, the Database property should be immutable' )
except AttributeError:
pass
except Exception, error:
self.fail( 'Modifying the Database property after connection is establilshed should raise AttributeError, bu %s was raised instead:\n %s' % ( error.__class__.__name__, error ) )
def testHost( self ):
"""Unit-tests the Host property of the MySQLConnector class."""
testObject = self.getLiveConnection()
oldHost = testObject.Host
testObject.Host = 'ook'
self.assertEquals( testObject.Host, 'ook', 'Before a connection is made, the object should allow modification of the Host property.' )
testObject.Host = oldHost
testObject.Connect()
try:
testObject.Host = 'ook'
self.fail( 'Once a connection is established, the Host property should be immutable' )
except AttributeError:
pass
except Exception, error:
self.fail( 'Modifying the Host property after connection is establilshed should raise AttributeError, bu %s was raised instead:\n %s' % ( error.__class__.__name__, error ) )
def testPassword( self ):
"""Unit-tests the Password property of the MySQLConnector class."""
testObject = self.getLiveConnection()
oldPassword = testObject.Password
testObject.Password = 'ook'
self.assertEquals( testObject.Password, 'ook', 'Before a connection is made, the object should allow modification of the Password property.' )
testObject.Password = oldPassword
testObject.Connect()
try:
testObject.Password = 'ook'
self.fail( 'Once a connection is established, the Password property should be immutable' )
except AttributeError:
pass
except Exception, error:
self.fail( 'Modifying the Password property after connection is establilshed should raise AttributeError, bu %s was raised instead:\n %s' % ( error.__class__.__name__, error ) )
# def testQueue( self ):
# """Unit-tests the Queue property of the MySQLConnector class."""
# pass
def testUser( self ):
"""Unit-tests the User property of the MySQLConnector class."""
testObject = self.getLiveConnection()
oldUser = testObject.User
testObject.User = 'ook'
self.assertEquals( testObject.User, 'ook', 'Before a connection is made, the object should allow modification of the User property.' )
testObject.User = oldUser
testObject.Connect()
try:
testObject.User = 'ook'
self.fail( 'Once a connection is established, the User property should be immutable' )
except AttributeError:
pass
except Exception, error:
self.fail( 'Modifying the User property after connection is establilshed should raise AttributeError, bu %s was raised instead:\n %s' % ( error.__class__.__name__, error ) )
# Unit-test methods
def testConfigure( self ):
"""Unit-tests the Configure method of the MySQLConnector class"""
def testConfigure( self ):
"""Unit-tests the Configure method of the BaseDataConnector abstract class."""
# First, we need to build out a configuration-file
hostName = 'localhost'
databaseName = 'test_database'
userName = 'test_user'
userPassword = 'test_password'
configString = """[Default Section]
SectionName1: SectionValue1
SectionName2: SectionValue2
[Data Access]
host: %s
database: %s
user: %s
password: %s
""" % ( hostName, databaseName, userName, userPassword )
fp = open( 'test.cfg', 'w' )
fp.write( configString )
fp.close()
# Now we need to create the configuration object and read in the configuration values
configParser = Configuration( 'test.cfg' )
# Now, finally, we can test the configuration
testObject = MySQLConnector()
testObject.Configure( configParser, 'Data Access' )
self.assertEquals( testObject.Host, hostName, 'The host name should be retrievable from configuration' )
self.assertEquals( testObject.Database, databaseName, 'The database name should be retrievable from configuration' )
self.assertEquals( testObject.User, userName, 'The user name should be retrievable from configuration' )
self.assertEquals( testObject.Password, userPassword, 'The user password should be retrievable from configuration' )
os.unlink( 'test.cfg' )
def testConnect( self ):
"""Unit-tests the Connect method of the MySQLConnector class."""
pass # Tested in the various property-tests above...
def testExecute( self ):
"""Unit-tests the Execute method of the MySQLConnector class."""
testObject = self.getLiveConnection()
# Single-table, single row, single field
MyQuery = Query( testObject, "SELECT first_name FROM test_table_1 LIMIT 1;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1, '%s should return a result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 1, '%s should return a single row' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), [ 'first_name' ], '%s should return only the first_name field' % ( MyQuery.Sql ) )
# Single-table, single row, all fields
MyQuery = Query( testObject, "SELECT * FROM test_table_1 LIMIT 1;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1, '%s should return a result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 1, '%s should return a single row' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0][0] ), 3, '%s should return three fields' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), [ 'first_name', 'last_name', 'id' ], '%s should return first_name, last_name and id fields' % ( MyQuery.Sql ) )
# Single-table, all rows, one field
MyQuery = Query( testObject, "SELECT first_name FROM test_table_1;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1, '%s should return a result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 2 )
self.assertEquals( testResults[0][0].keys(), [ 'first_name' ] )
self.assertEquals( testResults[0][0].keys(), testResults[0][1].keys(), 'Field-names in different rows of the result-set should be identical' )
# Single-table, all rows, all fields
MyQuery = Query( testObject, "SELECT * FROM test_table_1;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1, '%s should return a result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 2, '%s should return two rows' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0][0] ), 3, '%s should return three fields' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), [ 'first_name', 'last_name', 'id' ], '%s should return first_name, last_name and id fields' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), testResults[0][1].keys(), 'Field-names in different rows of the result-set should be identical' )
# Multiple tables (joined), all fields
MyQuery = Query( testObject, "SELECT t1.*, t2.* FROM test_table_1 t1 LEFT JOIN test_table_2 t2 ON t1.id=t2.name_id;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1, '%s should return a result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 2, '%s should return two rows' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), ['first_name', 'last_name', 't2.id', 'email', 'name_id', 'id'] )
self.assertEquals( testResults[0][0].keys(), testResults[0][1].keys(), 'Field-names in different rows of the result-set should be identical' )
# Multiple tables, multiple result-sets
MyQuery = Query( testObject, "SELECT * FROM test_table_1;SELECT * FROM test_table_2;" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 2, '%s should return two result-sets' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[0] ), 2, '%s should return two rows in it\'s first result-set' % ( MyQuery.Sql ) )
self.assertEquals( len( testResults[1] ), 2, '%s should return two rows in it\'s second result-set' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), [ 'first_name', 'last_name', 'id' ], '%s should return first_name, last_name and id fields' % ( MyQuery.Sql ) )
self.assertEquals( testResults[0][0].keys(), testResults[0][1].keys(), 'Field-names in different rows of the result-set should be identical' )
self.assertEquals( testResults[1][0].keys(), ['name_id', 'id', 'email'], '%s should return name_id, id and email fields' % ( MyQuery.Sql ) )
self.assertEquals( testResults[1][0].keys(), testResults[1][1].keys(), 'Field-names in different rows of the result-set should be identical' )
# Query that returns no results
MyQuery = Query( testObject, "SELECT * FROM test_table_1 WHERE first_name='ook';" )
testResults = testObject.Execute( MyQuery )
self.assertEquals( len( testResults ), 1 )
self.assertEquals( len( testResults[0] ), 0 )
testSuite.addTests( unittest.TestLoader().loadTestsFromTestCase( testMySQLConnector ) )
- Line(s)
- 1-8
- An almost-typical derived-test class. The main difference here is the inclusion of an explicit object-destructor that cannot throw errors.
- Unit-testing the object-destructor in any sort of meaningful way has eluded me still - I don't like not having such a test, but I'm recdonciled to having to take it on faith that the destructor is doing what it's supposed to do...
- 19-21
- A convenience method to return a valid
MySQLConnectorinstance pointing to the test database. - 23-31
- Standard final-class testing.
- 33-49
- Object-construction testing. Note that the properties being passed as
arguments to the constructor are being explicitly verified, though they
shouldn't have to be as long as
MySQLConnectorremains true to the interface ofBaseDataConnector. These are in place as much to make sure that potential future breaking changes underneath theMySQLConnectorclass would be caught before introducing bugs to a production system.
With the exception of the testConfigure and testExecute
methods, most of the rest of the tests provided here are pretty typical. The
testConfigure method is almost identical to the same method in the
test-suite for BaseDataConnector, but is provided as a separate
test-set in case of future modification to the Configure method of
either BaseDataConnector or MySQLConnector.
testExecute, on the other hand, is an entirely different flavor
of unicorn. This is really more of a system/integration test than a "real"
unit-test, though I think it may straddle the line to some degree. It's necessary
because the Execute method of MySQLConnector exists,
but at the same time, there's no good way to really test it without
making a connection to the database, and running queries against that connection.
I'll hammer out some additional (and more detailed) integration tests that will
eventually be added to the test-suite for MySQLConnector, but for now,
this feels pretty good: Given the two tables of two rows present in the test
database, I'm pretty confident that I've hit all of the meaningful variations
of number of results, rows and fields that could come up against that data-set.
No comments:
Post a Comment