connectivity/workben/postgresql/preparedstatement.py

   1 #*************************************************************************
   2 #
   3 #   $RCSfile: preparedstatement.py,v $
   4 #
   5 #   $Revision: 1.1.2.9 $
   6 #
   7 #   last change: $Author: jbu $ $Date: 2008/07/07 21:37: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: Ralph Thomas
  52 #
  53 #   Copyright: 2000 by Sun Microsystems, Inc.
  54 #
  55 #   All Rights Reserved.
  56 #
  57 #   Contributor(s): Ralph Thomas, Joerg Budischewski
  58 #
  59 #*************************************************************************
  60 import unittest
  61 import sys
  62 import ddl
  63 from uno import ByteSequence
  64 from com.sun.star.sdbc import SQLException
  65 from com.sun.star.sdbc.ResultSetConcurrency import UPDATABLE
  66 from com.sun.star.sdbc.DataType import NUMERIC,VARCHAR
  67
  68 def suite(ctx,dburl):
  69     suite = unittest.TestSuite()
  70     suite.addTest(TestCase("testQuery",ctx,dburl))
  71     suite.addTest(TestCase("testGeneratedResultSet",ctx,dburl))
  72     suite.addTest(TestCase("testUpdateableResultSet",ctx,dburl))
  73     suite.addTest(TestCase("testQuoteQuote",ctx,dburl))
  74     return suite
  75
  76 def realEquals( a,b,eps ):
  77     val = a - b
  78     if val < 0:
  79        val = -1. * val
  80     return val < eps
  81
  82 class TestCase(unittest.TestCase):
  83       def __init__(self,method,ctx,dburl):
  84           unittest.TestCase.__init__(self,method)
  85           self.ctx = ctx
  86           self.dburl = dburl
  87
  88       def setUp(self):
  89           self.driver = self.ctx.ServiceManager.createInstanceWithContext(
  90                    'org.openoffice.comp.connectivity.pq.Driver', self.ctx )
  91           self.connection = self.driver.connect( self.dburl, () )
  92           ddl.executeDDLs( self.connection )
  93
  94       def testDown( self ):
  95           self.connection.close()
  96
  97       def testQuery( self ):
  98
  99           stmts = "SELECT product.id FROM product WHERE product.price > :lowprice AND product.price < :upprice", \
 100               "SELECT product.id FROM product WHERE product.price > ? AND product.price < ?" , \
 101               "SELECT \"product\".\"id\" FROM product WHERE \"product\".\"price\" > :lowprice AND \"product\".\"price\" < :upprice"
 102
 103
 104           for stmt in stmts:
 105               prepstmt = self.connection.prepareStatement( stmt )
 106               prepstmt.setDouble( 1, 5.80 )
 107               prepstmt.setObjectWithInfo( 2, 7. , NUMERIC, 2)
 108               prepstmt.setObjectWithInfo( 2, "7.0000", NUMERIC, 2 )
 109               rs = prepstmt.executeQuery( )
 110               self.failUnless( rs.getMetaData().getColumnCount() == 1 )
 111               self.failUnless( rs.getMetaData().getColumnName(1) == "id")
 112               self.failUnless( prepstmt.getMetaData().getColumnCount() == 1 )
 113               self.failUnless( prepstmt.getMetaData().getColumnName(1) == "id" )
 114               self.failUnless( rs.next() )
 115               self.failUnless( rs.getString( 1 ).strip() == "PZZ2" )
 116               self.failUnless( rs.next() )
 117               self.failUnless( rs.getString( 1 ).strip() == "PZZ5" )
 118               self.failUnless( rs.isLast() )
 119
 120           prepstmt = self.connection.prepareStatement(
 121               "SELECT name FROM product WHERE id = ?" )
 122           prepstmt.setString( 1, 'PZZ2' )
 123           rs = prepstmt.executeQuery()
 124           self.failUnless( rs.next() )
 125           self.failUnless( rs.getString( 1 ) == "Pizza Mista" )
 126           self.failUnless( rs.isLast() )
 127
 128           prepstmt = self.connection.prepareStatement(
 129               "SELECT name FROM product WHERE image = ?" )
 130           prepstmt.setBytes( 1, ByteSequence( "\001foo\005" ) )
 131           rs = prepstmt.executeQuery()
 132           self.failUnless( rs.next() )
 133           self.failUnless( rs.getString( 1 ) == "Pizza Funghi" )
 134           self.failUnless( rs.isLast() )
 135
 136           prepstmt = self.connection.prepareStatement(
 137               "SELECT * FROM ordertab WHERE delivered = ?" )
 138           prepstmt.setBoolean( 1 , False )
 139           rs = prepstmt.executeQuery()
 140           self.failUnless( rs.next() )
 141           self.failUnless( rs.getString( 1 ).strip() == "2" )
 142           self.failUnless( rs.isLast() )
 143
 144           stmt = self.connection.createStatement()
 145           rs =  stmt.executeQuery( "SELECT * FROM \"public\".\"customer\"" )
 146
 147           stmt.executeUpdate( "DELETE FROM product where id='PAS5'" )
 148           prepstmt =self.connection.prepareStatement(
 149               "INSERT INTO product VALUES(?,'Ravioli',?,NULL)" );
 150           prepstmt.setObjectWithInfo( 1, "PAS5" ,VARCHAR,0)
 151           prepstmt.setObjectWithInfo( 2, "9.223" ,NUMERIC,2)
 152           prepstmt.executeUpdate()
 153           rs= stmt.executeQuery( "SELECT price FROM product WHERE id = 'PAS5'" )
 154           self.failUnless( rs.next() )
 155           self.failUnless( rs.getString( 1 ).strip() == "9.22" )
 156
 157           stmt.executeUpdate( "DELETE FROM product where id='PAS5'" )
 158           prepstmt =self.connection.prepareStatement(
 159               "INSERT INTO product VALUES('PAS5','Ravioli',?,NULL)" );
 160           prepstmt.setObjectWithInfo( 1, 9.223,NUMERIC,2 )
 161           prepstmt.executeUpdate()
 162           rs= stmt.executeQuery( "SELECT price FROM product WHERE id = 'PAS5'" )
 163           self.failUnless( rs.next() )
 164           self.failUnless( rs.getString( 1 ).strip() == "9.22" )
 165
 166       def testGeneratedResultSet( self ):
 167           prepstmt = self.connection.prepareStatement(
 168               "INSERT INTO customer VALUES( ?, ? )" )
 169           prepstmt.setString( 1, "C3" )
 170           prepstmt.setString( 2, "Norah Jones" )
 171           prepstmt.executeUpdate()
 172           rs = prepstmt.getGeneratedValues()
 173           self.failUnless( rs.next() )
 174           self.failUnless( rs.getInt( 3 ) == 3 )
 175
 176           prepstmt = self.connection.prepareStatement(
 177               "INSERT INTO public.nooid (id,name) VALUES( ?, ? )" )
 178           prepstmt.setString( 1, "C3" )
 179           prepstmt.setString( 2, "Norah Jones" )
 180           prepstmt.executeUpdate()
 181           rs = prepstmt.getGeneratedValues()
 182           self.failUnless( rs.next() )
 183           self.failUnless( rs.getString(1).rstrip() == "C3" )
 184
 185           prepstmt = self.connection.prepareStatement(
 186               "INSERT INTO public.nooid2 (name) VALUES( ? )" )
 187           prepstmt.setString( 1, "Norah Jones" )
 188           prepstmt.executeUpdate()
 189           rs = prepstmt.getGeneratedValues()
 190           self.failUnless( rs )
 191           self.failUnless( rs.next() )
 192           self.failUnless( rs.getString(2) == "Norah Jones" )
 193           self.failUnless( rs.getString(1) == "1" )
 194
 195       def testUpdateableResultSet( self ):
 196           stmt = self.connection.createStatement()
 197           stmt.ResultSetConcurrency = UPDATABLE
 198           rs = stmt.executeQuery( "SELECT * FROM orderpos" )
 199 #          ddl.dumpResultSet( rs )
 200           rs.next()
 201           rs.deleteRow()
 202           rs.next()
 203           rs.updateInt( 4 , 32 )
 204           rs.updateRow()
 205
 206           rs.moveToInsertRow()
 207           rs.updateString( 1 , '2' )
 208           rs.updateString( 2, '003' )
 209           rs.updateString( 3, 'PZZ5' )
 210           rs.updateInt( 4, 22 )
 211           rs.insertRow()
 212
 213           rs = stmt.executeQuery( "SELECT * FROM orderpos" )
 214           rs = stmt.executeQuery( "SELECT * FROM \"public\".\"orderpos\"" )
 215 #          ddl.dumpResultSet( rs )
 216
 217       def testQuoteQuote( self ):
 218           stmt = self.connection.prepareStatement( "select 'foo''l'" )
 219           rs = stmt.executeQuery()
 220           self.failUnless( rs )
 221           self.failUnless( rs.next() )
 222           self.failUnless( rs.getString(1) == "foo'l" )
 223
 224           stmt = self.connection.prepareStatement( "select 'foo''''l'" )
 225           rs = stmt.executeQuery()
 226           self.failUnless( rs )
 227           self.failUnless( rs.next() )
 228           self.failUnless( rs.getString(1) == "foo''l" )
 229
 230