connectivity/workben/postgresql/statement.py

   1 #*************************************************************************
   2 #
   3 #   $RCSfile: statement.py,v $
   4 #
   5 #   $Revision: 1.1.2.5 $
   6 #
   7 #   last change: $Author: jbu $ $Date: 2006/05/27 11:33:11 $
   8 #
   9 #   The Contents of this file are made available subject to the terms of
  10 #   either of the following licenses
  11 #
  12 #          - GNU Lesser General Public License Version 2.1
  13 #          - Sun Industry Standards Source License Version 1.1
  14 #
  15 #   Sun Microsystems Inc., October, 2000
  16 #
  17 #   GNU Lesser General Public License Version 2.1
  18 #   =============================================
  19 #   Copyright 2000 by Sun Microsystems, Inc.
  20 #   901 San Antonio Road, Palo Alto, CA 94303, USA
  21 #
  22 #   This library is free software; you can redistribute it and/or
  23 #   modify it under the terms of the GNU Lesser General Public
  24 #   License version 2.1, as published by the Free Software Foundation.
  25 #
  26 #   This library is distributed in the hope that it will be useful,
  27 #   but WITHOUT ANY WARRANTY; without even the implied warranty of
  28 #   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  29 #   Lesser General Public License for more details.
  30 #
  31 #   You should have received a copy of the GNU Lesser General Public
  32 #   License along with this library; if not, write to the Free Software
  33 #   Foundation, Inc., 59 Temple Place, Suite 330, Boston,
  34 #   MA  02111-1307  USA
  35 #
  36 #
  37 #   Sun Industry Standards Source License Version 1.1
  38 #   =================================================
  39 #   The contents of this file are subject to the Sun Industry Standards
  40 #   Source License Version 1.1 (the "License"); You may not use this file
  41 #   except in compliance with the License. You may obtain a copy of the
  42 #   License at http://www.openoffice.org/license.html.
  43 #
  44 #   Software provided under this License is provided on an "AS IS" basis,
  45 #   WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING,
  46 #   WITHOUT LIMITATION, WARRANTIES THAT THE SOFTWARE IS FREE OF DEFECTS,
  47 #   MERCHANTABLE, FIT FOR A PARTICULAR PURPOSE, OR NON-INFRINGING.
  48 #   See the License for the specific provisions governing your rights and
  49 #   obligations concerning the Software.
  50 #
  51 #   The Initial Developer of the Original Code is: Joerg Budischewski
  52 #
  53 #   Copyright: 2000 by Sun Microsystems, Inc.
  54 #
  55 #   All Rights Reserved.
  56 #
  57 #   Contributor(s): Joerg Budischewski
  58 #
  59 #
  60 #
  61 #*************************************************************************
  62 import unohelper
  63 import unittest
  64 import ddl
  65 from com.sun.star.sdbc import SQLException, XArray
  66 from com.sun.star.sdbc.DataType import VARCHAR
  67 from com.sun.star.util import Date
  68 from com.sun.star.util import Time
  69 from com.sun.star.util import DateTime
  70
  71 # todo
  72 class MyArray( unohelper.Base, XArray ):
  73     def __init__( self, data ):
  74         self.data = data
  75     def getBaseType( self ):
  76         return VARCHAR
  77     def getBaseTypeName( self ):
  78         return "varchar"
  79     def getArray( self, foo ):
  80         return self.data
  81     def getArrayAtIndex( self, startIndex, count, foo ):
  82         return self.data[startIndex:startIndex+count-1]
  83     def getResultSet( self, foo):
  84         return None
  85     def getResultSetAtIndex( self, startIndex, count, foo ):
  86         return None
  87
  88 def suite(ctx,dburl):
  89     suite = unittest.TestSuite()
  90     suite.addTest(TestCase("testRobustness",ctx,dburl))
  91     suite.addTest(TestCase("testRow",ctx,dburl))
  92     suite.addTest(TestCase("testNavigation",ctx,dburl))
  93     suite.addTest(TestCase("testDatabaseMetaData",ctx,dburl))
  94     suite.addTest(TestCase("testGeneratedResultSet",ctx,dburl))
  95     suite.addTest(TestCase("testResultSetMetaData",ctx,dburl))
  96     suite.addTest(TestCase("testArray",ctx,dburl))
  97     return suite
  98
  99 def realEquals( a,b,eps ):
 100     val = a - b
 101     if val < 0:
 102        val = -1. * val
 103     return val < eps
 104
 105 class TestCase(unittest.TestCase):
 106       def __init__(self,method,ctx,dburl):
 107           unittest.TestCase.__init__(self,method)
 108           self.ctx = ctx
 109           self.dburl = dburl
 110
 111       def setUp(self):
 112           self.driver = self.ctx.ServiceManager.createInstanceWithContext(
 113                    'org.openoffice.comp.connectivity.pq.Driver' , self.ctx )
 114           self.connection = self.driver.connect( self.dburl, () )
 115           self.stmt = self.connection.createStatement()
 116           try:
 117                 self.stmt.executeUpdate( "DROP TABLE firsttable" )
 118           except SQLException,e:
 119               pass
 120
 121           ddls = (
 122                "BEGIN",
 123                "CREATE TABLE firsttable (tString text,tInteger integer,tShort smallint,tLong bigint,tFloat real,"+
 124                        "tDouble double precision,tByteSeq bytea,tBool boolean, tDate date, tTime time, tTimestamp timestamp, tIntArray integer[], tStringArray text[], tSerial serial ) WITH OIDS",
 125                "INSERT INTO firsttable VALUES ( 'foo', 70000, 12000, 70001, 2.4, 2.45, 'huhu', 'true',  '1999-01-08','04:05:06','1999-01-08 04:05:06', '{2,3,4}', '{\"huhu\",\"hi\"}')",
 126                "INSERT INTO firsttable VALUES ( 'foo2', 69999, 12001, 70002, -2.4, 2.55, 'huhu', 'false', '1999-01-08','04:05:06','1999-01-08 04:05:06', NULL , '{\"bla\"}' )",
 127                "INSERT INTO firsttable VALUES ( 'foo2', 69999, 12001, 70002, -2.4, 2.55, 'huhu', null,  '1999-01-08', '04:05:06','1999-01-08 04:05:06', '{}' , '{\"bl ubs\",\"bl\\\\\\\\a}}b\\\\\"a\",\"blub\"}' )",
 128                "COMMIT" )
 129           for i in ddls:
 130               self.stmt.executeUpdate(i)
 131
 132       def tearDown(self):
 133           self.stmt.close()
 134           self.connection.close()
 135
 136       def testRow(self):
 137           row = ("foo",70000,12000,70001,2.4,2.45, "huhu", True ,
 138                  Date(8,1,1999), Time(0,6,5,4),DateTime(0,6,5,4,8,1,1999) )
 139           row2 = ("foo2",69999,12001,70002,-2.4,2.55, "huhu", False )
 140
 141           rs = self.stmt.executeQuery( "SELECT * from firsttable" )
 142           self.failUnless( rs.next() )
 143
 144           self.failUnless( rs.getString(1) == row[0] )
 145           self.failUnless( rs.getInt(2) == row[1] )
 146           self.failUnless( rs.getShort(3) == row[2] )
 147           self.failUnless( rs.getLong(4) == row[3] )
 148           self.failUnless( realEquals(rs.getFloat(5), row[4], 0.001))
 149           self.failUnless( realEquals(rs.getDouble(6), row[5], 0.00001))
 150           self.failUnless( rs.getBytes(7) == row[6] )
 151           self.failUnless( rs.getBoolean(8) == row[7] )
 152           self.failUnless( rs.getDate(9) == row[8] )
 153           self.failUnless( rs.getTime(10) == row[9] )
 154           self.failUnless( rs.getTimestamp(11) == row[10] )
 155
 156           a = rs.getArray(12)
 157           data = a.getArray( None )
 158           self.failUnless( len( data ) == 3 )
 159           self.failUnless( int(data[0] ) == 2 )
 160           self.failUnless( int(data[1] ) == 3 )
 161           self.failUnless( int(data[2] ) == 4 )
 162
 163           self.failUnless( rs.next() )
 164
 165           self.failUnless( rs.next() )
 166           data = rs.getArray(13).getArray(None)
 167           self.failUnless( data[0] == "bl ubs" )
 168           self.failUnless( data[1] == "bl\\a}}b\"a" ) # check special keys
 169           self.failUnless( data[2] == "blub" )
 170
 171           rs.getString(8)
 172           self.failUnless(  rs.wasNull()  )
 173           rs.getString(7)
 174           self.failUnless( not rs.wasNull() )
 175
 176           self.failUnless( rs.findColumn( "tShort" ) == 3 )
 177           rs.close()
 178
 179       def testNavigation( self ):
 180           rs = self.stmt.executeQuery( "SELECT * from firsttable" )
 181           self.failUnless( rs.isBeforeFirst() )
 182           self.failUnless( not rs.isAfterLast() )
 183           self.failUnless( rs.isBeforeFirst() )
 184
 185           self.failUnless( rs.next() )
 186           self.failUnless( rs.isFirst() )
 187           self.failUnless( not rs.isLast() )
 188           self.failUnless( not rs.isBeforeFirst() )
 189
 190           self.failUnless( rs.next() )
 191           self.failUnless( rs.next() )
 192           self.failUnless( not rs.next() )
 193           self.failUnless( rs.isAfterLast() )
 194
 195           rs.absolute( 1 )
 196           self.failUnless( rs.isFirst() )
 197
 198           rs.absolute( 3 )
 199           self.failUnless( rs.isLast() )
 200
 201           rs.relative( -1 )
 202           self.failUnless( rs.getRow() == 2 )
 203
 204           rs.relative( 1 )
 205           self.failUnless( rs.getRow() == 3 )
 206
 207           rs.close()
 208
 209       def testRobustness( self ):
 210           rs = self.stmt.executeQuery( "SELECT * from firsttable" )
 211
 212           self.failUnlessRaises( SQLException, rs.getString , 1 )
 213
 214           rs.next()
 215           self.failUnlessRaises( SQLException, rs.getString , 24 )
 216           self.failUnlessRaises( SQLException, rs.getString , 0 )
 217
 218           self.connection.close()
 219           self.failUnlessRaises( SQLException, rs.getString , 1 )
 220           self.failUnlessRaises( SQLException, self.stmt.executeQuery, "SELECT * from firsttable" )
 221           rs.close()
 222
 223
 224       def testDatabaseMetaData( self ):
 225           meta = self.connection.getMetaData()
 226
 227           self.failUnless( not meta.isReadOnly() )
 228
 229       def testGeneratedResultSet( self ):
 230           self.stmt.executeUpdate(
 231               "INSERT INTO firsttable VALUES ( 'foo3', 69998, 12001, 70002, -2.4, 2.55, 'huhu2')" )
 232           #ddl.dumpResultSet( self.stmt.getGeneratedValues() )
 233           rs = self.stmt.getGeneratedValues()
 234           self.failUnless( rs.next() )
 235           self.failUnless( rs.getInt( 14 )  == 4 )
 236
 237       def testResultSetMetaData( self ):
 238           rs = self.stmt.executeQuery( "SELECT * from firsttable" )
 239
 240           # just check, if we get results !
 241           meta = rs.getMetaData()
 242
 243           count = meta.getColumnCount()
 244           for i in range( 1, count+1):
 245               meta.isNullable( i )
 246               meta.isCurrency( i )
 247               meta.isCaseSensitive( i )
 248               meta.isSearchable( i )
 249               meta.isSigned( i )
 250               meta.getColumnDisplaySize( i )
 251               meta.getColumnName( i )
 252               meta.getColumnLabel( i )
 253               meta.getSchemaName( i )
 254               meta.getPrecision( i )
 255               meta.getScale( i )
 256               meta.getTableName( i )
 257               meta.getColumnTypeName( i )
 258               meta.getColumnType( i )
 259               meta.isReadOnly( i )
 260               meta.isWritable( i )
 261               meta.isDefinitelyWritable( i )
 262               meta.getColumnServiceName( i )
 263
 264       def testArray( self ):
 265           stmt = self.connection.prepareStatement(
 266               "INSERT INTO firsttable VALUES ( 'insertedArray', 70000, 12000, 70001, 2.4, 2.45, 'huhu', 'true',  '1999-01-08','04:05:06','1999-01-08 04:05:06', '{2,3,4}', ? )" )
 267           myarray = ( "a", "\"c", "}d{" )
 268           stmt.setArray( 1, MyArray( myarray ) )
 269           stmt.executeUpdate()
 270
 271           stmt = self.connection.createStatement()
 272           rs = stmt.executeQuery( "SELECT tStringArray FROM firsttable WHERE tString = 'insertedArray'" )
 273           rs.next()
 274           data =  rs.getArray(1).getArray(None)
 275           self.failUnless( data[0] == myarray[0] )
 276           self.failUnless( data[1] == myarray[1] )
 277           self.failUnless( data[2] == myarray[2] )
 278
 279