-
Notifications
You must be signed in to change notification settings - Fork 196
/
Copy pathtest_createDropDB.py
113 lines (104 loc) · 4.3 KB
/
test_createDropDB.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
#
# Licensed Materials - Property of IBM
#
# (c) Copyright IBM Corp. 2007-2008
#
from __future__ import print_function
import os
import sys
import unittest
import ibm_db
import config
from testfunctions import IbmDbTestFunctions
class IbmDbTestCase(unittest.TestCase):
@unittest.skipIf(os.environ.get("CI", False), "Test fails in CI")
# Fails with
# [IBM][CLI Driver][DB2/LINUXX8664] SQL0001N Binding or precompilation
# did not complete successfully. SQLCODE=-1
def test_createDropDB(self):
obj = IbmDbTestFunctions()
if ((obj.server.DBMS_NAME == "DB2") or (obj.server.DBMS_NAME[0:3] != "DB2")):
raise unittest.SkipTest("createdb, dropdb not Supported")
obj.assert_expect(self.run_test_createDropDB)
def run_test_createDropDB(self):
database = 'test001'
conn_str = "DATABASE=%s;HOSTNAME=%s;PORT=%d;PROTOCOL=TCPIP;UID=%s;PWD=%s;" % (database, config.hostname, config.port, config.user, config.password)
conn_str_attach = "attach=true;HOSTNAME=%s;PORT=%d;PROTOCOL=TCPIP;UID=%s;PWD=%s;" % (config.hostname, config.port, config.user, config.password) #for create db or drop db API it is nessesory that connection only attach to the DB server not to any existing database of DB server
conn_attach = ibm_db.connect(conn_str_attach, '', '')
if conn_attach:
conn = False
try:
conn = ibm_db.connect(conn_str, '', '')
except:
pass
if conn:
ibm_db.close(conn)
conn = False
try:
ibm_db.dropdb(conn_attach, database)
except:
print('Errors occurred during drop database')
try:
#create databse
rc = ibm_db.createdb(conn_attach, database)
if rc:
conn = ibm_db.connect(conn_str, '', '')
if conn:
print('database created sucessfully')
ibm_db.close(conn)
conn = False
else:
print('database is not created')
else:
print('Errors occurred during create database')
#drop databse
rc = ibm_db.dropdb(conn_attach, database)
if rc:
try:
conn = ibm_db.connect(conn_str, '', '')
except:
print('datbase droped sucessfully')
if conn:
print('Errors occurred during drop database')
ibm_db.close(conn)
conn = False
else:
print('Errors occurred during delete database')
#create database with codeset option
rc = ibm_db.createdb(conn_attach, database, 'iso88591')
if rc:
conn = ibm_db.connect(conn_str, '', '')
server_info = ibm_db.server_info( conn )
if conn and (server_info.DB_CODEPAGE == 819):
print('database with codeset created sucessfully')
ibm_db.close(conn)
conn = False
else:
print('database is not created')
else:
print('Errors occurred during create database')
#drop database
rc = ibm_db.dropdb(conn_attach, database)
if rc:
try:
conn = ibm_db.connect(conn_str, '', '')
except:
print('datbase droped sucessfully')
if conn:
print('Errors occurred during drop database')
ibm_db.close(conn)
conn = False
else:
print('Errors occurred during drop database')
except:
print(ibm_db.conn_errormsg())
pass
ibm_db.close(conn_attach)
else:
print(ibm_db.conn_errormsg())
#__END__
#__LUW_EXPECTED__
#database created sucessfully
#datbase droped sucessfully
#database with codeset created sucessfully
#datbase droped sucessfully