connectivity/workben/postgresql/sdbcx.py

   1 #*************************************************************************
   2 #
   3 #   $RCSfile: sdbcx.py,v $
   4 #
   5 #   $Revision: 1.1.2.6 $
   6 #
   7 #   last change: $Author: jbu $ $Date: 2007/01/07 13:50:38 $
   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 unittest
  63 import ddl
  64 import unohelper
  65 import sys
  66 from com.sun.star.sdbc import SQLException
  67 from com.sun.star.sdbc.DataType import VARCHAR, CHAR, DECIMAL, DOUBLE, BIGINT, NUMERIC
  68 from com.sun.star.sdbc.ColumnValue import NO_NULLS, NULLABLE
  69 from com.sun.star.sdbcx.KeyType import PRIMARY, FOREIGN, UNIQUE
  70 from com.sun.star.sdbc.KeyRule import RESTRICT, CASCADE, NO_ACTION
  71
  72 def suite(ctx,dburl):
  73     suite = unittest.TestSuite()
  74     suite.addTest(TestCase("testTables",ctx,dburl))
  75     suite.addTest(TestCase("testViews",ctx,dburl))
  76     suite.addTest(TestCase("testKeys",ctx,dburl))
  77     suite.addTest(TestCase("testUsers",ctx,dburl))
  78     suite.addTest(TestCase("testIndexes",ctx,dburl))
  79     return suite
  80
  81 def nullable2Str( v ):
  82     if v == NO_NULLS:
  83         return "NOT NULL"
  84     return ""
  85
  86 def autoIncremtent2Str( v ):
  87     if v:
  88         return "auto increment"
  89     return ""
  90
  91 def dumpColumns( columns ):
  92     n = columns.getCount()
  93     print "Name\t type\t prec\t scale\t"
  94     for i in range( 0, n ):
  95         col = columns.getByIndex( i )
  96         print col.Name + "\t "+col.TypeName + "\t " + str(col.Precision) + "\t " + str(col.Scale) + "\t "+\
  97               str( col.DefaultValue ) + "\t " + str( col.Description ) + "\t " +\
  98               autoIncremtent2Str( col.IsAutoIncrement ) + "\t " + \
  99               nullable2Str( col.IsNullable )
 100
 101
 102 class TestCase(unittest.TestCase):
 103       def __init__(self,method,ctx,dburl):
 104           unittest.TestCase.__init__(self,method)
 105           self.ctx = ctx
 106           self.dburl = dburl
 107
 108       def setUp( self ):
 109           self.driver = self.ctx.ServiceManager.createInstanceWithContext(
 110                    'org.openoffice.comp.connectivity.pq.Driver' , self.ctx )
 111           self.connection = self.driver.connect( self.dburl, () )
 112           ddl.executeDDLs( self.connection )
 113
 114       def tearDown( self ):
 115           self.connection.close()
 116
 117       def checkDescriptor( self, descriptor, name, typeName, type, prec, scale, defaultValue, desc ):
 118           self.failUnless( descriptor.Name == name )
 119           self.failUnless( descriptor.TypeName == typeName )
 120           self.failUnless( descriptor.Type == type )
 121           self.failUnless( descriptor.Precision == prec )
 122           self.failUnless( descriptor.Scale == scale )
 123 #          print descriptor.DefaultValue + " == " + defaultValue
 124 #          self.failUnless( descriptor.DefaultValue == defaultValue )
 125           self.failUnless( descriptor.Description == desc )
 126
 127
 128       def testKeys( self ):
 129           dd = self.driver.getDataDefinitionByConnection( self.connection )
 130           tables = dd.getTables()
 131           t = tables.getByName( "public.ordertab" )
 132           keys = t.getKeys()
 133           key = keys.getByName( "cust" )
 134           self.failUnless( key.Name == "cust" )
 135           self.failUnless( key.Type == FOREIGN )
 136           self.failUnless( key.ReferencedTable == "public.customer" )
 137           self.failUnless( key.UpdateRule == RESTRICT )
 138           self.failUnless( key.DeleteRule == CASCADE )
 139
 140           keycolumns = keys.getByName( "ordertab_pkey" ).getColumns()
 141           self.failUnless( keycolumns.getElementNames() == (u"id",) )
 142
 143           key = keys.getByName( "ordertab_pkey" )
 144           self.failUnless( key.Name == "ordertab_pkey" )
 145           self.failUnless( key.Type == PRIMARY )
 146           self.failUnless( key.UpdateRule == NO_ACTION )
 147           self.failUnless( key.DeleteRule == NO_ACTION )
 148
 149           keys = tables.getByName( "public.customer" ).getKeys()
 150           key = keys.getByName( "customer_dummyserial_key" )
 151           self.failUnless( key.Name == "customer_dummyserial_key" )
 152           self.failUnless( key.Type == UNIQUE )
 153           self.failUnless( key.UpdateRule == NO_ACTION )
 154           self.failUnless( key.DeleteRule == NO_ACTION )
 155
 156           keys = tables.getByName( "public.orderpos" ).getKeys()
 157           keyEnum = keys.createEnumeration()
 158           while keyEnum.hasMoreElements():
 159               key = keyEnum.nextElement()
 160               cols = key.getColumns()
 161               colEnum = cols.createEnumeration()
 162               while colEnum.hasMoreElements():
 163                   col = colEnum.nextElement()
 164
 165       def testViews( self ):
 166           dd = self.driver.getDataDefinitionByConnection( self.connection )
 167           views = dd.getViews()
 168
 169           v = views.getByName( "public.customer2" )
 170           self.failUnless( v.Name == "customer2" )
 171           self.failUnless( v.SchemaName == "public" )
 172           self.failUnless( v.Command != "" )
 173
 174       def testIndexes( self ):
 175           dd = self.driver.getDataDefinitionByConnection( self.connection )
 176           tables = dd.getTables()
 177           t = tables.getByName( "public.ordertab" )
 178           indexes = t.getIndexes()
 179           index = indexes.getByName( "ordertab_pkey" )
 180
 181           self.failUnless( index.Name == "ordertab_pkey" )
 182           self.failUnless( index.IsPrimaryKeyIndex )
 183           self.failUnless( index.IsUnique )
 184           self.failUnless( not index.IsClustered )
 185
 186           columns = index.getColumns()
 187           self.failUnless( columns.hasByName( "id" ) )
 188
 189           self.failUnless( columns.getByIndex(0).Name == "id" )
 190
 191       def checkRenameTable( self, t , tables):
 192           t.rename( "foo" )
 193           self.failUnless( tables.hasByName( "public.foo" ) )
 194
 195           t.rename( "public.foo2" )
 196           self.failUnless( tables.hasByName( "public.foo2" ) )
 197
 198           try:
 199               t.rename( "pqsdbc_test.foo2" )
 200               self.failUnless( tables.hasByName( "pqsdbc_test.foo2" ) )
 201               print "looks like a server 8.1 or later (changing a schema succeeded)"
 202               t.rename( "pqsdbc_test.foo" )
 203               self.failUnless( tables.hasByName( "pqsdbc_test.foo" ) )
 204               t.rename( "public.foo2" )
 205               self.failUnless( tables.hasByName( "public.foo2" ) )
 206           except SQLException,e:
 207               if e.Message.find( "support changing" ) >= 0:
 208                   print "looks like a server prior to 8.1 (changing schema failed with Message [" + e.Message.replace("\n", " ") + "])"
 209               else:
 210                   raise e
 211           tables.dropByName( "public.foo2" )
 212
 213       def testTables( self ):
 214           dd = self.driver.getDataDefinitionByConnection( self.connection )
 215           tables = dd.getTables()
 216           t = tables.getByName( "public.customer" )
 217           self.failUnless( t.Name == "customer" )
 218           self.failUnless( t.SchemaName == "public" )
 219           self.failUnless( t.Type == "TABLE" )
 220
 221           cols = t.getColumns()
 222           self.failUnless( cols.hasByName( 'name' ) )
 223           self.failUnless( cols.hasByName( 'id' ) )
 224           col = cols.getByName( "dummyserial" )
 225 #          dumpColumns( cols )
 226           self.checkDescriptor( cols.getByName( "id" ), "id", "bpchar", CHAR, 8, 0, "", "unique id" )
 227           self.checkDescriptor( cols.getByName( "name" ), "name", "text", VARCHAR, 0, 0, "", "" )
 228
 229           dd = cols.createDataDescriptor()
 230           dd.Name = "foo"
 231           dd.TypeName = "CHAR"
 232           dd.Type = CHAR
 233           dd.Precision = 25
 234           dd.IsNullable = NULLABLE
 235           cols.appendByDescriptor( dd )
 236
 237           dd.Name = "foo2"
 238           dd.TypeName = "DECIMAL"
 239           dd.Type = DECIMAL
 240           dd.Precision = 12
 241           dd.Scale = 5
 242           dd.DefaultValue = "2.3423"
 243           dd.Description = "foo2 description"
 244           cols.appendByDescriptor( dd )
 245
 246           dd.Name = "cash"
 247           dd.TypeName = "MONEY"
 248           dd.Type = DOUBLE
 249 #          dd.IsNullable = NO_NULLS
 250           dd.DefaultValue = "'2.42'"
 251           cols.appendByDescriptor( dd )
 252
 253           cols.refresh()
 254
 255           self.checkDescriptor( cols.getByName( "foo"), "foo", "bpchar", CHAR, 25,0,"","")
 256           self.checkDescriptor(
 257               cols.getByName( "foo2"), "foo2", "numeric", NUMERIC, 12,5,"2.3423","foo2 description")
 258 #          dumpColumns( cols )
 259
 260           datadesc = tables.createDataDescriptor()
 261           datadesc.SchemaName = "public"
 262           datadesc.Name = "blub"
 263           datadesc.Description = "This describes blub"
 264
 265           tables.appendByDescriptor( datadesc )
 266
 267           # make the appended descriptors known
 268           tables.refresh()
 269
 270           t = tables.getByName( "public.blub" )
 271           self.failUnless( t.Name == "blub" )
 272           self.failUnless( t.SchemaName == "public" )
 273           self.failUnless( t.Description == "This describes blub" )
 274
 275           cols = t.getColumns()
 276           dd = cols.createDataDescriptor()
 277           dd.Name = "mytext"
 278           dd.TypeName = "text"
 279           dd.Type = VARCHAR
 280           dd.IsNullable = NO_NULLS
 281           cols.appendByDescriptor( dd )
 282
 283           cols.refresh()
 284
 285           dd.DefaultValue = "'myDefault'"
 286           dd.Name = "mytext2"
 287           dd.IsNullable = NULLABLE
 288           dd.Description = "mytext-Description"
 289           t.alterColumnByName( "mytext" , dd )
 290
 291           cols.refresh()
 292
 293           self.checkDescriptor( cols.getByName( "mytext2" ), "mytext2", "text", VARCHAR, 0,0,"'myDefault'","mytext-Description" )
 294
 295           t = tables.getByName( "public.customer2" )
 296           self.checkRenameTable( t,tables )
 297
 298           t = tables.getByName( "public.blub" )
 299           self.checkRenameTable( t,tables )
 300
 301
 302
 303       def testUsers( self ):
 304           dd = self.driver.getDataDefinitionByConnection( self.connection )
 305           users = dd.getUsers()
 306           self.failUnless( "pqsdbc_joe" in users.getElementNames() )