11
11
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
12
# See the License for the specific language governing permissions and
13
13
# limitations under the License.
14
+ from typing import Optional
14
15
15
16
from robot .api import logger
16
17
@@ -20,7 +21,7 @@ class Assertion:
20
21
Assertion handles all the assertions of Database Library.
21
22
"""
22
23
23
- def check_if_exists_in_database (self , selectStatement , sansTran = False , msg = None ):
24
+ def check_if_exists_in_database (self , selectStatement : str , sansTran : bool = False , msg : Optional [ str ] = None ):
24
25
"""
25
26
Check if any row would be returned by given the input `selectStatement`. If there are no results, then this will
26
27
throw an AssertionError. Set optional input `sansTran` to True to run command without an explicit transaction
@@ -50,7 +51,7 @@ def check_if_exists_in_database(self, selectStatement, sansTran=False, msg=None)
50
51
msg or f"Expected to have have at least one row, " f"but got 0 rows from: '{ selectStatement } '"
51
52
)
52
53
53
- def check_if_not_exists_in_database (self , selectStatement , sansTran = False , msg = None ):
54
+ def check_if_not_exists_in_database (self , selectStatement : str , sansTran : bool = False , msg : Optional [ str ] = None ):
54
55
"""
55
56
This is the negation of `check_if_exists_in_database`.
56
57
@@ -77,13 +78,13 @@ def check_if_not_exists_in_database(self, selectStatement, sansTran=False, msg=N
77
78
| Check If Not Exists In Database | SELECT id FROM person WHERE first_name = 'Franz Allan' | msg=my error message |
78
79
"""
79
80
logger .info (f"Executing : Check If Not Exists In Database | { selectStatement } " )
80
- queryResults = self .query (selectStatement , sansTran )
81
- if queryResults :
81
+ query_results = self .query (selectStatement , sansTran )
82
+ if query_results :
82
83
raise AssertionError (
83
- msg or f"Expected to have have no rows from '{ selectStatement } ', but got some rows: { queryResults } "
84
+ msg or f"Expected to have have no rows from '{ selectStatement } ', but got some rows: { query_results } "
84
85
)
85
86
86
- def row_count_is_0 (self , selectStatement , sansTran = False , msg = None ):
87
+ def row_count_is_0 (self , selectStatement : str , sansTran : bool = False , msg : Optional [ str ] = None ):
87
88
"""
88
89
Check if any rows are returned from the submitted `selectStatement`. If there are, then this will throw an
89
90
AssertionError. Set optional input `sansTran` to True to run command without an explicit transaction commit or
@@ -107,12 +108,14 @@ def row_count_is_0(self, selectStatement, sansTran=False, msg=None):
107
108
Using optional `msg` to override the default error message:
108
109
| Row Count is 0 | SELECT id FROM person WHERE first_name = 'Franz Allan' | msg=my error message |
109
110
"""
110
- logger .info (f"Executing : Row Count Is 0 | selectStatement" )
111
+ logger .info (f"Executing : Row Count Is 0 | { selectStatement } " )
111
112
num_rows = self .row_count (selectStatement , sansTran )
112
113
if num_rows > 0 :
113
114
raise AssertionError (msg or f"Expected 0 rows, but { num_rows } were returned from: '{ selectStatement } '" )
114
115
115
- def row_count_is_equal_to_x (self , selectStatement , numRows , sansTran = False , msg = None ):
116
+ def row_count_is_equal_to_x (
117
+ self , selectStatement : str , numRows : str , sansTran : bool = False , msg : Optional [str ] = None
118
+ ):
116
119
"""
117
120
Check if the number of rows returned from `selectStatement` is equal to the value submitted. If not, then this
118
121
will throw an AssertionError. Set optional input `sansTran` to True to run command without an explicit
@@ -144,7 +147,9 @@ def row_count_is_equal_to_x(self, selectStatement, numRows, sansTran=False, msg=
144
147
msg or f"Expected { numRows } rows, but { num_rows } were returned from: '{ selectStatement } '"
145
148
)
146
149
147
- def row_count_is_greater_than_x (self , selectStatement , numRows , sansTran = False , msg = None ):
150
+ def row_count_is_greater_than_x (
151
+ self , selectStatement : str , numRows : str , sansTran : bool = False , msg : Optional [str ] = None
152
+ ):
148
153
"""
149
154
Check if the number of rows returned from `selectStatement` is greater than the value submitted. If not, then
150
155
this will throw an AssertionError. Set optional input `sansTran` to True to run command without an explicit
@@ -176,7 +181,9 @@ def row_count_is_greater_than_x(self, selectStatement, numRows, sansTran=False,
176
181
msg or f"Expected more than { numRows } rows, but { num_rows } were returned from '{ selectStatement } '"
177
182
)
178
183
179
- def row_count_is_less_than_x (self , selectStatement , numRows , sansTran = False , msg = None ):
184
+ def row_count_is_less_than_x (
185
+ self , selectStatement : str , numRows : str , sansTran : bool = False , msg : Optional [str ] = None
186
+ ):
180
187
"""
181
188
Check if the number of rows returned from `selectStatement` is less than the value submitted. If not, then this
182
189
will throw an AssertionError. Set optional input `sansTran` to True to run command without an explicit
@@ -208,7 +215,7 @@ def row_count_is_less_than_x(self, selectStatement, numRows, sansTran=False, msg
208
215
msg or f"Expected less than { numRows } rows, but { num_rows } were returned from '{ selectStatement } '"
209
216
)
210
217
211
- def table_must_exist (self , tableName , sansTran = False , msg = None ):
218
+ def table_must_exist (self , tableName : str , sansTran : bool = False , msg : Optional [ str ] = None ):
212
219
"""
213
220
Check if the table given exists in the database. Set optional input `sansTran` to True to run command without an
214
221
explicit transaction commit or rollback. The default error message can be overridden with the `msg` argument.
@@ -228,33 +235,31 @@ def table_must_exist(self, tableName, sansTran=False, msg=None):
228
235
Using optional `msg` to override the default error message:
229
236
| Table Must Exist | first_name | msg=my error message |
230
237
"""
231
- logger .info ("Executing : Table Must Exist | %s " % tableName )
238
+ logger .info (f "Executing : Table Must Exist | { tableName } " )
232
239
if self .db_api_module_name in ["cx_Oracle" , "oracledb" ]:
233
- selectStatement = (
240
+ query = (
234
241
"SELECT * FROM all_objects WHERE object_type IN ('TABLE','VIEW') AND "
235
- "owner = SYS_CONTEXT('USERENV', 'SESSION_USER') AND object_name = UPPER('%s ')" % tableName
242
+ f "owner = SYS_CONTEXT('USERENV', 'SESSION_USER') AND object_name = UPPER('{ tableName } ')"
236
243
)
237
- table_exists = self .row_count (selectStatement , sansTran ) > 0
244
+ table_exists = self .row_count (query , sansTran ) > 0
238
245
elif self .db_api_module_name in ["sqlite3" ]:
239
- selectStatement = (
240
- "SELECT name FROM sqlite_master WHERE type='table' AND name='%s' COLLATE NOCASE" % tableName
241
- )
242
- table_exists = self .row_count (selectStatement , sansTran ) > 0
246
+ query = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{ tableName } ' COLLATE NOCASE"
247
+ table_exists = self .row_count (query , sansTran ) > 0
243
248
elif self .db_api_module_name in ["ibm_db" , "ibm_db_dbi" ]:
244
- selectStatement = "SELECT name FROM SYSIBM.SYSTABLES WHERE type='T' AND name=UPPER('%s ')" % tableName
245
- table_exists = self .row_count (selectStatement , sansTran ) > 0
249
+ query = f "SELECT name FROM SYSIBM.SYSTABLES WHERE type='T' AND name=UPPER('{ tableName } ')"
250
+ table_exists = self .row_count (query , sansTran ) > 0
246
251
elif self .db_api_module_name in ["teradata" ]:
247
- selectStatement = "SELECT TableName FROM DBC.TablesV WHERE TableKind='T' AND TableName='%s'" % tableName
248
- table_exists = self .row_count (selectStatement , sansTran ) > 0
252
+ query = f "SELECT TableName FROM DBC.TablesV WHERE TableKind='T' AND TableName='{ tableName } '"
253
+ table_exists = self .row_count (query , sansTran ) > 0
249
254
else :
250
255
try :
251
- selectStatement = f"SELECT * FROM information_schema.tables WHERE table_name='{ tableName } '"
252
- table_exists = self .row_count (selectStatement , sansTran ) > 0
256
+ query = f"SELECT * FROM information_schema.tables WHERE table_name='{ tableName } '"
257
+ table_exists = self .row_count (query , sansTran ) > 0
253
258
except :
254
259
logger .info ("Database doesn't support information schema, try using a simple SQL request" )
255
260
try :
256
- selectStatement = f"SELECT 1 from { tableName } where 1=0"
257
- num_rows = self .row_count (selectStatement , sansTran )
261
+ query = f"SELECT 1 from { tableName } where 1=0"
262
+ self .row_count (query , sansTran )
258
263
table_exists = True
259
264
except :
260
265
table_exists = False
0 commit comments