diff options
Diffstat (limited to 'connectivity/workben/postgresql/preparedstatement.py')
| -rw-r--r-- | connectivity/workben/postgresql/preparedstatement.py | 228 |
1 files changed, 228 insertions, 0 deletions
diff --git a/connectivity/workben/postgresql/preparedstatement.py b/connectivity/workben/postgresql/preparedstatement.py new file mode 100644 index 000000000000..d049c9f6f5c3 --- /dev/null +++ b/connectivity/workben/postgresql/preparedstatement.py @@ -0,0 +1,228 @@ +#************************************************************************* +# +# $RCSfile: preparedstatement.py,v $ +# +# $Revision: 1.1.2.9 $ +# +# last change: $Author: jbu $ $Date: 2008/07/07 21:37:11 $ +# +# The Contents of this file are made available subject to the terms of +# either of the following licenses +# +# - GNU Lesser General Public License Version 2.1 +# - Sun Industry Standards Source License Version 1.1 +# +# Sun Microsystems Inc., October, 2000 +# +# GNU Lesser General Public License Version 2.1 +# ============================================= +# Copyright 2000 by Sun Microsystems, Inc. +# 901 San Antonio Road, Palo Alto, CA 94303, USA +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License version 2.1, as published by the Free Software Foundation. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA +# +# +# Sun Industry Standards Source License Version 1.1 +# ================================================= +# The contents of this file are subject to the Sun Industry Standards +# Source License Version 1.1 (the "License"); You may not use this file +# except in compliance with the License. You may obtain a copy of the +# License at http://www.openoffice.org/license.html. +# +# Software provided under this License is provided on an "AS IS" basis, +# WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, +# WITHOUT LIMITATION, WARRANTIES THAT THE SOFTWARE IS FREE OF DEFECTS, +# MERCHANTABLE, FIT FOR A PARTICULAR PURPOSE, OR NON-INFRINGING. +# See the License for the specific provisions governing your rights and +# obligations concerning the Software. +# +# The Initial Developer of the Original Code is: Ralph Thomas +# +# Copyright: 2000 by Sun Microsystems, Inc. +# +# All Rights Reserved. +# +# Contributor(s): Ralph Thomas, Joerg Budischewski +# +#************************************************************************* +import unittest +import sys +import ddl +from uno import ByteSequence +from com.sun.star.sdbc import SQLException +from com.sun.star.sdbc.ResultSetConcurrency import UPDATABLE +from com.sun.star.sdbc.DataType import NUMERIC,VARCHAR + +def suite(ctx,dburl): + suite = unittest.TestSuite() + suite.addTest(TestCase("testQuery",ctx,dburl)) + suite.addTest(TestCase("testGeneratedResultSet",ctx,dburl)) + suite.addTest(TestCase("testUpdateableResultSet",ctx,dburl)) + suite.addTest(TestCase("testQuoteQuote",ctx,dburl)) + return suite + +def realEquals( a,b,eps ): + val = a - b + if val < 0: + val = -1. * val + return val < eps + +class TestCase(unittest.TestCase): + def __init__(self,method,ctx,dburl): + unittest.TestCase.__init__(self,method) + self.ctx = ctx + self.dburl = dburl + + def setUp(self): + self.driver = self.ctx.ServiceManager.createInstanceWithContext( + 'org.openoffice.comp.connectivity.pq.Driver', self.ctx ) + self.connection = self.driver.connect( self.dburl, () ) + ddl.executeDDLs( self.connection ) + + def testDown( self ): + self.connection.close() + + def testQuery( self ): + + stmts = "SELECT product.id FROM product WHERE product.price > :lowprice AND product.price < :upprice", \ + "SELECT product.id FROM product WHERE product.price > ? AND product.price < ?" , \ + "SELECT \"product\".\"id\" FROM product WHERE \"product\".\"price\" > :lowprice AND \"product\".\"price\" < :upprice" + + + for stmt in stmts: + prepstmt = self.connection.prepareStatement( stmt ) + prepstmt.setDouble( 1, 5.80 ) + prepstmt.setObjectWithInfo( 2, 7. , NUMERIC, 2) + prepstmt.setObjectWithInfo( 2, "7.0000", NUMERIC, 2 ) + rs = prepstmt.executeQuery( ) + self.failUnless( rs.getMetaData().getColumnCount() == 1 ) + self.failUnless( rs.getMetaData().getColumnName(1) == "id") + self.failUnless( prepstmt.getMetaData().getColumnCount() == 1 ) + self.failUnless( prepstmt.getMetaData().getColumnName(1) == "id" ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ).strip() == "PZZ2" ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ).strip() == "PZZ5" ) + self.failUnless( rs.isLast() ) + + prepstmt = self.connection.prepareStatement( + "SELECT name FROM product WHERE id = ?" ) + prepstmt.setString( 1, 'PZZ2' ) + rs = prepstmt.executeQuery() + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ) == "Pizza Mista" ) + self.failUnless( rs.isLast() ) + + prepstmt = self.connection.prepareStatement( + "SELECT name FROM product WHERE image = ?" ) + prepstmt.setBytes( 1, ByteSequence( "\001foo\005" ) ) + rs = prepstmt.executeQuery() + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ) == "Pizza Funghi" ) + self.failUnless( rs.isLast() ) + + prepstmt = self.connection.prepareStatement( + "SELECT * FROM ordertab WHERE delivered = ?" ) + prepstmt.setBoolean( 1 , False ) + rs = prepstmt.executeQuery() + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ).strip() == "2" ) + self.failUnless( rs.isLast() ) + + stmt = self.connection.createStatement() + rs = stmt.executeQuery( "SELECT * FROM \"public\".\"customer\"" ) + + stmt.executeUpdate( "DELETE FROM product where id='PAS5'" ) + prepstmt =self.connection.prepareStatement( + "INSERT INTO product VALUES(?,'Ravioli',?,NULL)" ); + prepstmt.setObjectWithInfo( 1, "PAS5" ,VARCHAR,0) + prepstmt.setObjectWithInfo( 2, "9.223" ,NUMERIC,2) + prepstmt.executeUpdate() + rs= stmt.executeQuery( "SELECT price FROM product WHERE id = 'PAS5'" ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ).strip() == "9.22" ) + + stmt.executeUpdate( "DELETE FROM product where id='PAS5'" ) + prepstmt =self.connection.prepareStatement( + "INSERT INTO product VALUES('PAS5','Ravioli',?,NULL)" ); + prepstmt.setObjectWithInfo( 1, 9.223,NUMERIC,2 ) + prepstmt.executeUpdate() + rs= stmt.executeQuery( "SELECT price FROM product WHERE id = 'PAS5'" ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString( 1 ).strip() == "9.22" ) + + def testGeneratedResultSet( self ): + prepstmt = self.connection.prepareStatement( + "INSERT INTO customer VALUES( ?, ? )" ) + prepstmt.setString( 1, "C3" ) + prepstmt.setString( 2, "Norah Jones" ) + prepstmt.executeUpdate() + rs = prepstmt.getGeneratedValues() + self.failUnless( rs.next() ) + self.failUnless( rs.getInt( 3 ) == 3 ) + + prepstmt = self.connection.prepareStatement( + "INSERT INTO public.nooid (id,name) VALUES( ?, ? )" ) + prepstmt.setString( 1, "C3" ) + prepstmt.setString( 2, "Norah Jones" ) + prepstmt.executeUpdate() + rs = prepstmt.getGeneratedValues() + self.failUnless( rs.next() ) + self.failUnless( rs.getString(1).rstrip() == "C3" ) + + prepstmt = self.connection.prepareStatement( + "INSERT INTO public.nooid2 (name) VALUES( ? )" ) + prepstmt.setString( 1, "Norah Jones" ) + prepstmt.executeUpdate() + rs = prepstmt.getGeneratedValues() + self.failUnless( rs ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString(2) == "Norah Jones" ) + self.failUnless( rs.getString(1) == "1" ) + + def testUpdateableResultSet( self ): + stmt = self.connection.createStatement() + stmt.ResultSetConcurrency = UPDATABLE + rs = stmt.executeQuery( "SELECT * FROM orderpos" ) +# ddl.dumpResultSet( rs ) + rs.next() + rs.deleteRow() + rs.next() + rs.updateInt( 4 , 32 ) + rs.updateRow() + + rs.moveToInsertRow() + rs.updateString( 1 , '2' ) + rs.updateString( 2, '003' ) + rs.updateString( 3, 'PZZ5' ) + rs.updateInt( 4, 22 ) + rs.insertRow() + + rs = stmt.executeQuery( "SELECT * FROM orderpos" ) + rs = stmt.executeQuery( "SELECT * FROM \"public\".\"orderpos\"" ) +# ddl.dumpResultSet( rs ) + + def testQuoteQuote( self ): + stmt = self.connection.prepareStatement( "select 'foo''l'" ) + rs = stmt.executeQuery() + self.failUnless( rs ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString(1) == "foo'l" ) + + stmt = self.connection.prepareStatement( "select 'foo''''l'" ) + rs = stmt.executeQuery() + self.failUnless( rs ) + self.failUnless( rs.next() ) + self.failUnless( rs.getString(1) == "foo''l" ) |