3 Copyright (c) 2011-2014 ARM Limited
5 Licensed under the Apache License, Version 2.0 (the "License");
6 you may not use this file except in compliance with the License.
7 You may obtain a copy of the License at
9 http://www.apache.org/licenses/LICENSE-2.0
11 Unless required by applicable law or agreed to in writing, software
12 distributed under the License is distributed on an "AS IS" BASIS,
13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 See the License for the specific language governing permissions and
15 limitations under the License.
17 Author: Przemyslaw Wirkus <Przemyslaw.Wirkus@arm.com>
23 # Imports from TEST API
24 from workspace_tools.test_db import BaseDBAccess
27 class MySQLDBAccess(BaseDBAccess):
28 """ Wrapper for MySQL DB access for common test suite interface
31 BaseDBAccess.__init__(self)
32 self.DB_TYPE = 'mysql'
34 def detect_database(self, verbose=False):
35 """ detect database and return VERION data structure or string (verbose=True)
37 query = 'SHOW VARIABLES LIKE "%version%"'
38 rows = self.select_all(query)
42 result.append("\t%s: %s"% (row['Variable_name'], row['Value']))
43 result = "\n".join(result)
48 def parse_db_connection_string(self, str):
49 """ Parsing SQL DB connection string. String should contain:
50 - DB Name, user name, password, URL (DB host), name
51 Function should return tuple with parsed (host, user, passwd, db) or None if error
52 E.g. connection string: 'mysql://username:password@127.0.0.1/db_name'
54 result = BaseDBAccess().parse_db_connection_string(str)
55 if result is not None:
56 (db_type, username, password, host, db_name) = result
57 if db_type != 'mysql':
61 def is_connected(self):
62 """ Returns True if we are connected to database
64 return self.db_object is not None
66 def connect(self, host, user, passwd, db):
67 """ Connects to DB and returns DB object
70 self.db_object = mdb.connect(host=host, user=user, passwd=passwd, db=db)
71 # Let's remember connection credentials
72 self.db_type = self.DB_TYPE
78 print "Error %d: %s"% (e.args[0], e.args[1])
86 def connect_url(self, db_url):
87 """ Connects to database using db_url (database url parsing),
88 store host, username, password, db_name
90 result = self.parse_db_connection_string(db_url)
91 if result is not None:
92 (db_type, username, password, host, db_name) = result
93 if db_type == self.DB_TYPE:
94 self.connect(host, username, password, db_name)
97 """ Reconnects to DB and returns DB object using stored host name,
98 database name and credentials (user name and password)
100 self.connect(self.host, self.user, self.passwd, self.db)
102 def disconnect(self):
103 """ Close DB connection
106 self.db_object.close()
107 self.db_object = None
110 def escape_string(self, str):
111 """ Escapes string so it can be put in SQL query between quotes
114 result = con.escape_string(str)
115 return result if result else ''
117 def select_all(self, query):
118 """ Execute SELECT query and get all results
121 cur = con.cursor(mdb.cursors.DictCursor)
123 rows = cur.fetchall()
126 def insert(self, query, commit=True):
127 """ Execute INSERT query, define if you want to commit
136 def get_next_build_id(self, name, desc='', location='', type=None, status=None):
137 """ Insert new build_id (DB unique build like ID number to send all test results)
140 status = self.BUILD_ID_STATUS_STARTED
143 type = self.BUILD_ID_TYPE_TEST
145 query = """INSERT INTO `%s` (%s_name, %s_desc, %s_location, %s_type_fk, %s_status_fk)
146 VALUES ('%s', '%s', '%s', %d, %d)"""% (self.TABLE_BUILD_ID,
152 self.escape_string(name),
153 self.escape_string(desc),
154 self.escape_string(location),
157 index = self.insert(query) # Provide inserted record PK
160 def get_table_entry_pk(self, table, column, value, update_db=True):
161 """ Checks for entries in tables with two columns (<TABLE_NAME>_pk, <column>)
162 If update_db is True updates table entry if value in specified column doesn't exist
164 # TODO: table buffering
166 table_pk = '%s_pk'% table
167 query = """SELECT `%s`
169 WHERE `%s`='%s'"""% (table_pk,
172 self.escape_string(value))
173 rows = self.select_all(query)
175 result = rows[0][table_pk]
176 elif len(rows) == 0 and update_db:
177 # Update DB with new value
178 result = self.update_table_entry(table, column, value)
181 def update_table_entry(self, table, column, value):
182 """ Updates table entry if value in specified column doesn't exist
183 Locks table to perform atomic read + update
188 cur.execute("LOCK TABLES `%s` WRITE"% table)
189 table_pk = '%s_pk'% table
190 query = """SELECT `%s`
192 WHERE `%s`='%s'"""% (table_pk,
195 self.escape_string(value))
197 rows = cur.fetchall()
199 query = """INSERT INTO `%s` (%s)
200 VALUES ('%s')"""% (table,
202 self.escape_string(value))
204 result = cur.lastrowid
206 cur.execute("UNLOCK TABLES")
209 def update_build_id_info(self, build_id, **kw):
210 """ Update additional data inside build_id table
212 db.update_build_id_info(build_id, _status_fk=self.BUILD_ID_STATUS_COMPLETED, _shuffle_seed=0.0123456789):
217 # Prepare UPDATE query
218 # ["`mtest_build_id_pk`=[value-1]", "`mtest_build_id_name`=[value-2]", "`mtest_build_id_desc`=[value-3]"]
221 assign_str = "`%s%s`='%s'"% (self.TABLE_BUILD_ID, col_sufix, self.escape_string(str(kw[col_sufix])))
222 set_list.append(assign_str)
223 set_str = ', '.join(set_list)
224 query = """UPDATE `%s`
226 WHERE `mtest_build_id_pk`=%d"""% (self.TABLE_BUILD_ID,
232 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=''):
233 """ Inserts test result entry to database. All checks regarding existing
234 toolchain names in DB are performed.
235 If some data is missing DB will be updated
237 # Get all table FK and if entry is new try to insert new value
238 target_fk = self.get_table_entry_pk(self.TABLE_TARGET, self.TABLE_TARGET + '_name', target)
239 toolchain_fk = self.get_table_entry_pk(self.TABLE_TOOLCHAIN, self.TABLE_TOOLCHAIN + '_name', toolchain)
240 test_type_fk = self.get_table_entry_pk(self.TABLE_TEST_TYPE, self.TABLE_TEST_TYPE + '_name', test_type)
241 test_id_fk = self.get_table_entry_pk(self.TABLE_TEST_ID, self.TABLE_TEST_ID + '_name', test_id)
242 test_result_fk = self.get_table_entry_pk(self.TABLE_TEST_RESULT, self.TABLE_TEST_RESULT + '_name', test_result)
247 query = """ INSERT INTO `%s` (`mtest_build_id_fk`,
249 `mtest_toolchain_fk`,
250 `mtest_test_type_fk`,
252 `mtest_test_result_fk`,
255 `mtest_test_timeout`,
256 `mtest_test_loop_no`,
257 `mtest_test_result_extra`)
258 VALUES (%d, %d, %d, %d, %d, %d, '%s', %.2f, %.2f, %d, '%s')"""% (self.TABLE_TEST_ENTRY,
265 self.escape_string(test_output),
269 self.escape_string(test_extra))