Unit Testing

From Devipedia

Jump to: navigation, search

Contents

Jython

Uses Python syntax and Java JVM

Using JDBC

There are two options for using from Jython. One is using the native JAVA OracleDriver and DriverManager...

from oracle.jdbc.driver import OracleDriver
from java.sql import DriverManager

def connect(un, pw, sid, host, port):
     driver = OracleDriver()
     DriverManager.registerDriver(driver)
     connection = "jdbc:oracle:thin:@%s:%s:%s" % (host, port, sid)
     conn = DriverManager.getConnection(connection, un, pw)
     return conn

def doStuff(conn):
     stmt = conn.createStatement()
     rset = stmt.executeQuery("SELECT banner FROM sys.v_$version")
     while (rset.next()):
         print rset.getString(1)
     stmt.close()

if __name__ == "__main__":
     un = 'dvenable'
     pw = 'passwd'
     sid = 'mysid'
     host = 'myhost'
     port = '1521'
     conn = connect(un, pw, sid, host, port)
     doStuff(conn)

...the other is using the Python DB API 2.0 compliant zxJDBC...

from com.ziclix.python.sql import zxJDBC

url = 'jdbc:oracle:thin:@myhost:1521:mysid'
user = 'dvenable'
passwd = 'passwd'


db = zxJDBC.connect(url, user, passwd, "oracle.jdbc.OracleDriver")
c = db.cursor()
c.execute("SELECT banner FROM sys.v_$version")
for all in c.fetchall():
    print all


Concurrency

My goal is to use a unit testing framework to perform load tests. That means we need the ability to run concurrent tests. Here are a few approaches to threading. Use the python threading module...

import threading

class WorkerThread( threading.Thread ):

    def run (self):
        print "Here we go doing work in the thread..."
        print self.getName()


for i in range(1, 10):
    WorkerThread().start()

...or go native...

from java.lang import Thread, Runnable

class WorkerJob( Runnable ):
   def __init__( self, name ):
       self.name = name
   def run( self ):
       print self.name

for i in range(0,10):
    t = Thread( WorkerJob( "Runnable" ), "MyThread %d" % i)
    t.start()

Unit Testing Framework

Python and Jython have a built in unit testing framework based on Kent Beck's patterns. It works very much like JUnit, cppunit, etc.

import unittest
import random

class TestSequenceFunctions(unittest.TestCase):

    def setUp(self):
        self.seq = range(10)

    def testshuffle(self):
        # make sure the shuffled sequence does not lose any elements
        random.shuffle(self.seq)
        self.seq.sort()
        self.assertEqual(self.seq, range(10))

    def testchoice(self):
        element = random.choice(self.seq)
        self.assert_(element in self.seq)

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

Run as follows to see the output:

dvenable@dvenable:~/src/jython$ jython unit.py 
..
----------------------------------------------------------------------
Ran 2 tests in 0.001s

OK

In the above case, the main function (__main__) will execute unittest.main() which will automatically run all test methods on all classes that implement unittest.TestCase. Finer-grained control is available if needed and the common JUnit constructs (fixtures, suites, etc.) are all available.

import unittest
import random

class TestClassOne(unittest.TestCase):

    def testone(self):
        self.assertEqual(1, 1)

    def testtwo(self):
        self.assert_(1)

class TestClassTwo(unittest.TestCase):

    def testthree(self):
        self.assertEqual(1, 1)

    def testfour(self):
        self.assert_(1)

if __name__ == '__main__':

    widgetTestSuite = unittest.TestSuite()
    widgetTestSuite.addTest(TestClassOne('testone'))
    widgetTestSuite.addTest(TestClassOne('testtwo'))
    widgetTestSuite.addTest(TestClassTwo('testthree'))
    widgetTestSuite.addTest(TestClassTwo('testfour'))
    tr = unittest.TestResult()
    tr.shouldStop = True
    widgetTestSuite.run(tr)
    print "Errors: %d " % len(tr.errors)
    print "Failures: %d " % len(tr.failures)

Unit Test with Concurrency Example

import unittest
import random
import threading

class Shuffle( threading.Thread ):

    def __init__(self, unit):
        self.unit = unit
        threading.Thread.__init__(self)

    def run(self):
        # make sure the shuffled sequence does not lose any elements
        seq = range(10)
        seq.sort()
        self.unit.assertEqual(seq, range(10))
        # force error
        self.unit.assertEqual(1, 2)

class TestConcurrent(unittest.TestCase):

    def testshuffle(self):
        for i in range(1, 100):
            s = Shuffle(self)
            s.start()

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

Personal tools