229 lines
9.3 KiB
Python
229 lines
9.3 KiB
Python
#*************************************************************************
|
|
#
|
|
# $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" )
|