1
2
3 import MySQLdb
4 import warnings
5 import pytest
6
7
9 """
10 Connects to MySQL DB, executes queries
11 Autocommit is disabled by default, commit is done with commit function
12 Warnings are turned into exceptions
13 """
15 """
16 Initialize DB connection. Turning warning into exceptions
17 """
18 warnings.filterwarnings('error', category=MySQLdb.Warning)
19 self.db = None
20 try:
21 self.db = MySQLdb.connect(**config_data)
22 self.cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
23 self.db.autocommit(False)
24 except MySQLdb.Error, err:
25 print "Error %d: %s" % (err.args[0], err.args[1])
26 pytest.fail('FAILED connecting to DB with error: %d: %s' % (err.args[0], err.args[1]))
27
29 """
30 Returns status of autocommit LOCAL/GLOBAL/SESSION variables (local by default)
31 :return: tuple(dict) i.e. ({'Value': 'OFF', 'Variable_name': 'autocommit'},)
32 """
33 if (var_type in ('local', 'global', 'session')) is False:
34 pytest.fail("Wrong argument '%s' passed in <get_auto_commit> function. Valid args are str: 'local', 'global' or 'session'" % var_type)
35
36 status_query = "show " + var_type + " variables where variable_name = 'autocommit'"
37 return self.select_query(status_query)
38
40 """
41 Send/execute queries
42 :param: str MySQL query
43 :param: str 'one'/'all' Rows to be fetched
44 :return: fetch = one dictionary i.e. {row_1}
45 :return: fetch != one tuple with arguments dictionaries, each dictionary is a row i.e. ({row_first}, {row_second}, ... ,{row_last})
46 """
47 try:
48 self.cursor.execute(query)
49 data = self.cursor.fetchone() if fetch == 'one' else self.cursor.fetchall()
50 return data
51 except MySQLdb.Error, err:
52 print "Error %d: %s" % (err.args[0], err.args[1])
53 pytest.fail('<select_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
54
56 """
57 Sends query to DB.
58 :param: str query
59 :return: int Nr of rows updated (in case of updating with same value, query is successful but nr of rows updated = 0
60 """
61 try:
62 self.cursor.execute(query)
63 rows_updated = self.cursor.rowcount
64 print "Number of rows updated: %s" % rows_updated
65 except MySQLdb.Error, err:
66 print "Error %d: %s" % (err.args[0], err.args[1])
67 pytest.fail('<update_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
68
69 return rows_updated
70
72 """
73 to be updated: don't have permission to delete on current working DB
74 """
75 try:
76 self.cursor.execute(query)
77 rows_deleted = self.cursor.rowcount
78 print "Number of rows deleted: %s" % rows_deleted
79 except MySQLdb.Error, err:
80 print "Error %d: %s" % (err.args[0], err.args[1])
81 pytest.fail('<delete_query()> function FAILURE with error: %d: %s' % (err.args[0], err.args[1]))
82
83 return rows_deleted
84
87
90
93
96
98 self.cursor.close()
99 self.db.close()
100
101
102
103
104 config_file = {
105 'host': 'mtlcarqa01.itops.ad.ypg.com',
106 'user': 'cmsqauser',
107 'passwd': 'cmsqapass',
108 'db': 'cmspub'
109 }
110
111 db = MysqlConnector(config_file)
112 print db.get_autocommit_status()
113