<s> import os <EOL> import time <EOL> import subprocess <EOL> from twisted . internet import protocol , reactor <EOL> from twisted . python import log <EOL> from twisted . trial import unittest <EOL> from nagcat . plugins import query_oracle <EOL> from nagcat . unittests . queries import QueryTestCase <EOL> from nagcat import errors <EOL> from coil . struct import Struct <EOL> try : <EOL> import cx_Oracle <EOL> from lxml import etree <EOL> except ImportError : <EOL> cx_Oracle = None <EOL> etree = None <EOL> class OracleBase ( QueryTestCase ) : <EOL> if not cx_Oracle or not etree : <EOL> skip = "<STR_LIT>" <EOL> elif not ( '<STR_LIT>' in os . environ and <EOL> '<STR_LIT>' in os . environ and <EOL> '<STR_LIT>' in os . environ ) : <EOL> skip = "<STR_LIT>" <EOL> SQL_SETUP = ( ) <EOL> SQL_CLEAN = ( ) <EOL> QUERY_TYPE = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( OracleBase , self ) . setUp ( ) <EOL> self . config = { <EOL> '<STR_LIT:type>' : self . QUERY_TYPE , <EOL> '<STR_LIT:user>' : os . environ [ '<STR_LIT>' ] , <EOL> '<STR_LIT:password>' : os . environ [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : os . environ [ '<STR_LIT>' ] } <EOL> if self . SQL_SETUP : <EOL> self . execute ( self . SQL_SETUP ) <EOL> def tearDown ( self ) : <EOL> if self . SQL_CLEAN : <EOL> self . execute ( self . SQL_CLEAN ) <EOL> def execute ( self , sqlseq ) : <EOL> conn = cx_Oracle . Connection ( user = self . config [ '<STR_LIT:user>' ] , <EOL> password = self . config [ '<STR_LIT:password>' ] , <EOL> dsn = self . config [ '<STR_LIT>' ] , <EOL> threaded = True ) <EOL> self . execute_in_connection ( sqlseq , conn ) <EOL> conn . close ( ) <EOL> def execute_in_connection ( self , sqlseq , conn ) : <EOL> cursor = conn . cursor ( ) <EOL> for sql in sqlseq : <EOL> try : <EOL> cursor . execute ( sql ) <EOL> except cx_Oracle . DatabaseError , ex : <EOL> raise Exception ( "<STR_LIT>" % ( ex , sql ) ) <EOL> cursor . close ( ) <EOL> def startQuery ( self , ** kwargs ) : <EOL> return super ( OracleBase , self ) . startQuery ( self . config , ** kwargs ) <EOL> def assertEqualsXML ( self , result , expect ) : <EOL> parser = etree . XMLParser ( remove_blank_text = True ) <EOL> result = etree . tostring ( etree . XML ( result , parser ) ) <EOL> expect = etree . tostring ( etree . XML ( expect , parser ) ) <EOL> self . assertEquals ( result , expect ) <EOL> class SimpleTestCase ( OracleBase ) : <EOL> def testSimple ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> d = self . startQuery ( sql = '<STR_LIT>' ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testBinds ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> d = self . startQuery ( <EOL> sql = '<STR_LIT>' , <EOL> binds = [ <NUM_LIT:1> ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testParams1 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> d = self . startQuery ( <EOL> sql = '<STR_LIT>' , <EOL> parameters = [ <NUM_LIT:2> ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testParams2 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> d = self . startQuery ( <EOL> sql = '<STR_LIT>' , <EOL> parameters = Struct ( { '<STR_LIT>' : <NUM_LIT:2> } ) ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testString ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> d = self . startQuery ( sql = "<STR_LIT>" ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testBadQuery ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> d = self . startQuery ( sql = '<STR_LIT>' ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> def testBadUser ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> d = self . startQuery ( sql = '<STR_LIT>' , user = '<STR_LIT>' ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> class DataTestCase ( OracleBase ) : <EOL> SQL_SETUP = ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> SQL_CLEAN = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def testSelectAll ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , """<STR_LIT>""" ) <EOL> d = self . startQuery ( sql = '<STR_LIT>' ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testSelectCount ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , """<STR_LIT>""" ) <EOL> d = self . startQuery ( sql = '<STR_LIT>' ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testNonSelect ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , "<STR_LIT>" ) <EOL> d = self . startQuery ( sql = "<STR_LIT>" ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> class TimeoutQueryTestCase ( OracleBase ) : <EOL> SQL_SETUP = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> SQL_CLEAN = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> super ( TimeoutQueryTestCase , self ) . setUp ( ) <EOL> self . locked_conn = cx_Oracle . Connection ( <EOL> user = self . config [ '<STR_LIT:user>' ] , <EOL> password = self . config [ '<STR_LIT:password>' ] , <EOL> dsn = self . config [ '<STR_LIT>' ] , <EOL> threaded = True ) <EOL> self . execute_in_connection ( ( <EOL> "<STR_LIT>" , <EOL> ) , self . locked_conn ) <EOL> def tearDown ( self ) : <EOL> self . locked_conn . close ( ) <EOL> self . locked_conn = None <EOL> super ( TimeoutQueryTestCase , self ) . tearDown ( ) <EOL> def test_timeout ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> self . assert_ ( str ( result . value ) . startswith ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" % result . value ) <EOL> deferred = self . startQuery ( <EOL> sql = '<STR_LIT>' , <EOL> timeout = <NUM_LIT:0.5> ) <EOL> deferred . addBoth ( check ) <EOL> return deferred <EOL> class DummyFactory ( protocol . Factory ) : <EOL> protocol = protocol . Protocol <EOL> class TimeoutConnectionTestCase ( QueryTestCase ) : <EOL> """<STR_LIT>""" <EOL> if not cx_Oracle or not etree : <EOL> skip = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( TimeoutConnectionTestCase , self ) . setUp ( ) <EOL> self . server = reactor . listenTCP ( <NUM_LIT> , DummyFactory ( ) ) <EOL> self . config = { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:user>' : '<STR_LIT>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0.5> , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def tearDown ( self ) : <EOL> return self . server . stopListening ( ) <EOL> def test_timeout ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> self . assert_ ( str ( result . value ) . startswith ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" % result . value ) <EOL> d = self . startQuery ( self . config ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> class PLSQLTestCase ( OracleBase ) : <EOL> SQL_CLEAN = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> QUERY_TYPE = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( PLSQLTestCase , self ) . setUp ( ) <EOL> path = "<STR_LIT>" % ( os . path . dirname ( os . path . abspath ( __file__ ) ) , <EOL> "<STR_LIT>" ) <EOL> input = open ( path ) <EOL> proc = subprocess . Popen ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" % ( <EOL> self . config [ '<STR_LIT:user>' ] , <EOL> self . config [ '<STR_LIT:password>' ] , <EOL> self . config [ '<STR_LIT>' ] ) ] , <EOL> stdin = input , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT ) <EOL> input . close ( ) <EOL> out , bleh = proc . communicate ( ) <EOL> for line in out . splitlines ( ) : <EOL> line = line . strip ( ) <EOL> if line : <EOL> log . msg ( "<STR_LIT>" % line ) <EOL> assert proc . returncode == <NUM_LIT:0> <EOL> def test_1 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , <EOL> """<STR_LIT>""" ) <EOL> d = self . startQuery ( procedure = "<STR_LIT>" , <EOL> parameters = [ [ '<STR_LIT>' , '<STR_LIT>' , "<STR_LIT>" ] ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def test_2 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , <EOL> """<STR_LIT>""" ) <EOL> d = self . startQuery ( procedure = "<STR_LIT>" , <EOL> parameters = [ [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:7> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def test_3 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , <EOL> """<STR_LIT>""" ) <EOL> d = self . startQuery ( procedure = "<STR_LIT>" , <EOL> parameters = [ [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def test_4 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , <EOL> """<STR_LIT>""" ) <EOL> d = self . startQuery ( procedure = "<STR_LIT>" , <EOL> parameters = [ [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def test_5 ( self ) : <EOL> def check ( result ) : <EOL> self . assertEqualsXML ( result , <EOL> """<STR_LIT>""" ) <EOL> d = self . startQuery ( procedure = "<STR_LIT>" , <EOL> parameters = [ [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> class ForkItTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testNone ( self ) : <EOL> proc = query_oracle . ForkIt ( <NUM_LIT:1> , lambda : None ) <EOL> d = proc . getResult ( ) <EOL> d . addBoth ( self . assertIdentical , None ) <EOL> return d <EOL> def testTrue ( self ) : <EOL> proc = query_oracle . ForkIt ( <NUM_LIT:1> , lambda : True ) <EOL> d = proc . getResult ( ) <EOL> d . addBoth ( self . assertIdentical , True ) <EOL> return d <EOL> def testAbort ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> self . assertIsInstance ( result . value , errors . TestUnknown ) <EOL> self . assertIn ( "<STR_LIT>" , <EOL> str ( result . value ) ) <EOL> proc = query_oracle . ForkIt ( <NUM_LIT:1> , os . _exit , <NUM_LIT:0> ) <EOL> d = proc . getResult ( ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> testAbort . skip = "<STR_LIT>" <EOL> def testTimeout ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , errors . Failure ) <EOL> self . assertIsInstance ( result . value , errors . TestCritical ) <EOL> self . assertIn ( "<STR_LIT>" , str ( result . value ) ) <EOL> proc = query_oracle . ForkIt ( <NUM_LIT:0.1> , time . sleep , <NUM_LIT:10> ) <EOL> d = proc . getResult ( ) <EOL> d . addBoth ( check ) <EOL> return d </s>
