Marco Zecchini
/
Example_RTOS
Rtos API example
Embed:
(wiki syntax)
Show/hide line numbers
test_mysql.py
00001 """ 00002 mbed SDK 00003 Copyright (c) 2011-2014 ARM Limited 00004 00005 Licensed under the Apache License, Version 2.0 (the "License"); 00006 you may not use this file except in compliance with the License. 00007 You may obtain a copy of the License at 00008 00009 http://www.apache.org/licenses/LICENSE-2.0 00010 00011 Unless required by applicable law or agreed to in writing, software 00012 distributed under the License is distributed on an "AS IS" BASIS, 00013 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 00014 See the License for the specific language governing permissions and 00015 limitations under the License. 00016 00017 Author: Przemyslaw Wirkus <Przemyslaw.Wirkus@arm.com> 00018 """ 00019 00020 import re 00021 import MySQLdb as mdb 00022 00023 # Imports from TEST API 00024 from tools.test_db import BaseDBAccess 00025 00026 00027 class MySQLDBAccess (BaseDBAccess ): 00028 """ Wrapper for MySQL DB access for common test suite interface 00029 """ 00030 def __init__(self): 00031 BaseDBAccess.__init__(self) 00032 self.DB_TYPE = 'mysql' 00033 00034 def detect_database (self, verbose=False): 00035 """ detect database and return VERION data structure or string (verbose=True) 00036 """ 00037 query = 'SHOW VARIABLES LIKE "%version%"' 00038 rows = self.select_all (query) 00039 if verbose: 00040 result = [] 00041 for row in rows: 00042 result.append("\t%s: %s"% (row['Variable_name'], row['Value'])) 00043 result = "\n".join(result) 00044 else: 00045 result = rows 00046 return result 00047 00048 def parse_db_connection_string (self, str): 00049 """ Parsing SQL DB connection string. String should contain: 00050 - DB Name, user name, password, URL (DB host), name 00051 Function should return tuple with parsed (host, user, passwd, db) or None if error 00052 E.g. connection string: 'mysql://username:password@127.0.0.1/db_name' 00053 """ 00054 result = BaseDBAccess().parse_db_connection_string(str) 00055 if result is not None: 00056 (db_type, username, password, host, db_name) = result 00057 if db_type != 'mysql': 00058 result = None 00059 return result 00060 00061 def is_connected (self): 00062 """ Returns True if we are connected to database 00063 """ 00064 return self.db_object is not None 00065 00066 def connect (self, host, user, passwd, db): 00067 """ Connects to DB and returns DB object 00068 """ 00069 try: 00070 self.db_object = mdb.connect(host=host, user=user, passwd=passwd, db=db) 00071 # Let's remember connection credentials 00072 self.db_type = self.DB_TYPE 00073 self.host = host 00074 self.user = user 00075 self.passwd = passwd 00076 self.db = db 00077 except mdb.Error, e: 00078 print "Error %d: %s"% (e.args[0], e.args[1]) 00079 self.db_object = None 00080 self.db_type = None 00081 self.host = None 00082 self.user = None 00083 self.passwd = None 00084 self.db = None 00085 00086 def connect_url (self, db_url): 00087 """ Connects to database using db_url (database url parsing), 00088 store host, username, password, db_name 00089 """ 00090 result = self.parse_db_connection_string (db_url) 00091 if result is not None: 00092 (db_type, username, password, host, db_name) = result 00093 if db_type == self.DB_TYPE : 00094 self.connect (host, username, password, db_name) 00095 00096 def reconnect (self): 00097 """ Reconnects to DB and returns DB object using stored host name, 00098 database name and credentials (user name and password) 00099 """ 00100 self.connect (self.host , self.user , self.passwd , self.db ) 00101 00102 def disconnect (self): 00103 """ Close DB connection 00104 """ 00105 if self.db_object : 00106 self.db_object .close() 00107 self.db_object = None 00108 self.db_type = None 00109 00110 def escape_string (self, str): 00111 """ Escapes string so it can be put in SQL query between quotes 00112 """ 00113 con = self.db_object 00114 result = con.escape_string(str) 00115 return result if result else '' 00116 00117 def select_all (self, query): 00118 """ Execute SELECT query and get all results 00119 """ 00120 con = self.db_object 00121 cur = con.cursor(mdb.cursors.DictCursor) 00122 cur.execute(query) 00123 rows = cur.fetchall() 00124 return rows 00125 00126 def insert (self, query, commit=True): 00127 """ Execute INSERT query, define if you want to commit 00128 """ 00129 con = self.db_object 00130 cur = con.cursor() 00131 cur.execute(query) 00132 if commit: 00133 con.commit() 00134 return cur.lastrowid 00135 00136 def get_next_build_id (self, name, desc='', location='', type=None, status=None): 00137 """ Insert new build_id (DB unique build like ID number to send all test results) 00138 """ 00139 if status is None: 00140 status = self.BUILD_ID_STATUS_STARTED 00141 00142 if type is None: 00143 type = self.BUILD_ID_TYPE_TEST 00144 00145 query = """INSERT INTO `%s` (%s_name, %s_desc, %s_location, %s_type_fk, %s_status_fk) 00146 VALUES ('%s', '%s', '%s', %d, %d)"""% (self.TABLE_BUILD_ID, 00147 self.TABLE_BUILD_ID, 00148 self.TABLE_BUILD_ID, 00149 self.TABLE_BUILD_ID, 00150 self.TABLE_BUILD_ID, 00151 self.TABLE_BUILD_ID, 00152 self.escape_string (name), 00153 self.escape_string (desc), 00154 self.escape_string (location), 00155 type, 00156 status) 00157 index = self.insert (query) # Provide inserted record PK 00158 return index 00159 00160 def get_table_entry_pk (self, table, column, value, update_db=True): 00161 """ Checks for entries in tables with two columns (<TABLE_NAME>_pk, <column>) 00162 If update_db is True updates table entry if value in specified column doesn't exist 00163 """ 00164 # TODO: table buffering 00165 result = None 00166 table_pk = '%s_pk'% table 00167 query = """SELECT `%s` 00168 FROM `%s` 00169 WHERE `%s`='%s'"""% (table_pk, 00170 table, 00171 column, 00172 self.escape_string (value)) 00173 rows = self.select_all (query) 00174 if len(rows) == 1: 00175 result = rows[0][table_pk] 00176 elif len(rows) == 0 and update_db: 00177 # Update DB with new value 00178 result = self.update_table_entry (table, column, value) 00179 return result 00180 00181 def update_table_entry (self, table, column, value): 00182 """ Updates table entry if value in specified column doesn't exist 00183 Locks table to perform atomic read + update 00184 """ 00185 result = None 00186 con = self.db_object 00187 cur = con.cursor() 00188 cur.execute("LOCK TABLES `%s` WRITE"% table) 00189 table_pk = '%s_pk'% table 00190 query = """SELECT `%s` 00191 FROM `%s` 00192 WHERE `%s`='%s'"""% (table_pk, 00193 table, 00194 column, 00195 self.escape_string (value)) 00196 cur.execute(query) 00197 rows = cur.fetchall() 00198 if len(rows) == 0: 00199 query = """INSERT INTO `%s` (%s) 00200 VALUES ('%s')"""% (table, 00201 column, 00202 self.escape_string (value)) 00203 cur.execute(query) 00204 result = cur.lastrowid 00205 con.commit() 00206 cur.execute("UNLOCK TABLES") 00207 return result 00208 00209 def update_build_id_info (self, build_id, **kw): 00210 """ Update additional data inside build_id table 00211 Examples: 00212 db.update_build_id_info(build_id, _status_fk=self.BUILD_ID_STATUS_COMPLETED, _shuffle_seed=0.0123456789): 00213 """ 00214 if len(kw): 00215 con = self.db_object 00216 cur = con.cursor() 00217 # Prepare UPDATE query 00218 # ["`mtest_build_id_pk`=[value-1]", "`mtest_build_id_name`=[value-2]", "`mtest_build_id_desc`=[value-3]"] 00219 set_list = [] 00220 for col_sufix in kw: 00221 assign_str = "`%s%s`='%s'"% (self.TABLE_BUILD_ID, col_sufix, self.escape_string (str(kw[col_sufix]))) 00222 set_list.append(assign_str) 00223 set_str = ', '.join(set_list) 00224 query = """UPDATE `%s` 00225 SET %s 00226 WHERE `mtest_build_id_pk`=%d"""% (self.TABLE_BUILD_ID, 00227 set_str, 00228 build_id) 00229 cur.execute(query) 00230 con.commit() 00231 00232 def insert_test_entry (self, build_id, target, toolchain, test_type, test_id, test_result, test_output, test_time, test_timeout, test_loop, test_extra=''): 00233 """ Inserts test result entry to database. All checks regarding existing 00234 toolchain names in DB are performed. 00235 If some data is missing DB will be updated 00236 """ 00237 # Get all table FK and if entry is new try to insert new value 00238 target_fk = self.get_table_entry_pk (self.TABLE_TARGET, self.TABLE_TARGET + '_name', target) 00239 toolchain_fk = self.get_table_entry_pk (self.TABLE_TOOLCHAIN, self.TABLE_TOOLCHAIN + '_name', toolchain) 00240 test_type_fk = self.get_table_entry_pk (self.TABLE_TEST_TYPE, self.TABLE_TEST_TYPE + '_name', test_type) 00241 test_id_fk = self.get_table_entry_pk (self.TABLE_TEST_ID, self.TABLE_TEST_ID + '_name', test_id) 00242 test_result_fk = self.get_table_entry_pk (self.TABLE_TEST_RESULT, self.TABLE_TEST_RESULT + '_name', test_result) 00243 00244 con = self.db_object 00245 cur = con.cursor() 00246 00247 query = """ INSERT INTO `%s` (`mtest_build_id_fk`, 00248 `mtest_target_fk`, 00249 `mtest_toolchain_fk`, 00250 `mtest_test_type_fk`, 00251 `mtest_test_id_fk`, 00252 `mtest_test_result_fk`, 00253 `mtest_test_output`, 00254 `mtest_test_time`, 00255 `mtest_test_timeout`, 00256 `mtest_test_loop_no`, 00257 `mtest_test_result_extra`) 00258 VALUES (%d, %d, %d, %d, %d, %d, '%s', %.2f, %.2f, %d, '%s')"""% (self.TABLE_TEST_ENTRY, 00259 build_id, 00260 target_fk, 00261 toolchain_fk, 00262 test_type_fk, 00263 test_id_fk, 00264 test_result_fk, 00265 self.escape_string (test_output), 00266 test_time, 00267 test_timeout, 00268 test_loop, 00269 self.escape_string (test_extra)) 00270 cur.execute(query) 00271 con.commit()
Generated on Sun Jul 17 2022 08:25:32 by 1.7.2