diff options
author | Caolán McNamara <caolanm@redhat.com> | 2012-11-02 12:19:00 +0000 |
---|---|---|
committer | Caolán McNamara <caolanm@redhat.com> | 2012-11-02 12:19:00 +0000 |
commit | 3cdd4964f086f2c771237be7ad70b68f6e9e92cc (patch) | |
tree | 20f1666d2192c4973c12d576a09c2cb6a9952bb6 /connectivity/workben/postgresql/preparedstatement.py | |
parent | 3b6ba84a21f90bd1619999cf2445c62c71124513 (diff) |
ditch SISSL using workben code
Change-Id: I9a55180e0a0ef1bc14ce7c84265fb3d5508bd541
Diffstat (limited to 'connectivity/workben/postgresql/preparedstatement.py')
-rw-r--r-- | connectivity/workben/postgresql/preparedstatement.py | 222 |
1 files changed, 0 insertions, 222 deletions
diff --git a/connectivity/workben/postgresql/preparedstatement.py b/connectivity/workben/postgresql/preparedstatement.py deleted file mode 100644 index a44f2ae55472..000000000000 --- a/connectivity/workben/postgresql/preparedstatement.py +++ /dev/null @@ -1,222 +0,0 @@ -#************************************************************************* -# -# 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.noext', 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" ) |