ARMmbed
/
mbed-os
mirror of https://github.com/ARMmbed/mbed-os.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Database: added new factory and verbose database detection functions 

Added separate file to store test suite DB access interface

Added new functionality to database interface: database name, connection status, hostname + uname functionality

Added option --db to singletest.py so users can define database connection url for database access

Added database configuration check with switch --config. Now users can combine switch --db and --config to check database conenctivity
pull/475/head
Przemek Wirkus 2014-08-18 15:33:24 +01:00
parent 4aaf7d4e3f
commit 35fd4ad7d9
4 changed files with 206 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
workspace_tools/singletest.py
View File

@ -17,18 +17,10 @@ See the License for the specific language governing permissions and
limitations under the License.
Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
"""
-------------------------------------------------------------------------------
Call:
singletest.py --help
to get help information.
-------------------------------------------------------------------------------
File format example: test_spec.json
"""
File format example: test_spec.json:
{
"targets": {
"KL46Z": ["ARM", "GCC_ARM"],
@ -38,8 +30,7 @@ File format example: test_spec.json
}
}
File format example: muts_all.json
File format example: muts_all.json:
{
"1" : {"mcu": "LPC1768",
"port":"COM4",
@ -53,9 +44,9 @@ File format example: muts_all.json
"peripherals": ["digital_loop", "port_loop", "analog_loop"]
}
}
"""
# Check if 'prettytable' module is installed
try:
from prettytable import PrettyTable
@ -81,13 +72,16 @@ sys.path.insert(0, ROOT)
from workspace_tools.build_api import mcu_toolchain_matrix
# Imports from TEST API
from workspace_tools.test_api import BaseDBAccess
from workspace_tools.test_api import SingleTestRunner
from workspace_tools.test_api import factory_db_logger
from workspace_tools.test_api import singletest_in_cli_mode
from workspace_tools.test_api import detect_database_verbose
from workspace_tools.test_api import get_json_data_from_file
from workspace_tools.test_api import print_muts_configuration_from_json
from workspace_tools.test_api import print_test_configuration_from_json
from workspace_tools.test_api import get_avail_tests_summary_table
from workspace_tools.test_api import get_default_test_options_parser
from workspace_tools.test_api import print_muts_configuration_from_json
from workspace_tools.test_api import print_test_configuration_from_json
def get_version():
@ -119,6 +113,10 @@ if __name__ == '__main__':
print get_avail_tests_summary_table()
exit(0)
if opts.db_url and opts.verbose_test_configuration_only:
detect_database_verbose(opts.db_url)
exit(0)
# Print summary / information about automation test status
if opts.test_case_report:
test_case_report_cols = ['id', 'automated', 'description', 'peripherals', 'host_test', 'duration', 'source_dir']
@ -156,6 +154,7 @@ if __name__ == '__main__':
print print_test_configuration_from_json(test_spec)
exit(0)
# Verbose test specification and MUTs configuration
if MUTs and opts.verbose:
print print_muts_configuration_from_json(MUTs)
@ -169,6 +168,7 @@ if __name__ == '__main__':
single_test = SingleTestRunner(_global_loops_count=opts.test_global_loops_value,
_test_loops_list=opts.test_loops_list,
_muts=MUTs,
_opts_db_url=opts.db_url,
_opts_log_file_name=opts.log_file_name,
_test_spec=test_spec,
_opts_goanna_for_mbed_sdk=opts.goanna_for_mbed_sdk,

132
workspace_tools/test_api.py
View File

@ -45,6 +45,8 @@ from workspace_tools.targets import TARGET_MAP
from workspace_tools.build_api import build_project, build_mbed_libs, build_lib
from workspace_tools.build_api import get_target_supported_toolchains
from workspace_tools.libraries import LIBRARIES, LIBRARY_MAP
from workspace_tools.test_db import BaseDBAccess
from workspace_tools.test_mysql import MySQLDBAccess
class ProcessObserver(Thread):
@ -130,6 +132,7 @@ class SingleTestRunner(object):
_global_loops_count=1,
_test_loops_list=None,
_muts={},
_opts_db_url=None,
_opts_log_file_name=None,
_test_spec={},
_opts_goanna_for_mbed_sdk=None,
@ -173,6 +176,7 @@ class SingleTestRunner(object):
self.test_spec = _test_spec
# Settings passed e.g. from command line
self.opts_db_url = _opts_db_url
self.opts_log_file_name = _opts_log_file_name
self.opts_goanna_for_mbed_sdk = _opts_goanna_for_mbed_sdk
self.opts_goanna_for_tests = _opts_goanna_for_tests
@ -194,6 +198,7 @@ class SingleTestRunner(object):
self.opts_extend_test_timeout = _opts_extend_test_timeout
self.logger = CLITestLogger(file_name=self.opts_log_file_name) # Default test logger
self.db_logger = factory_db_logger(self.opts_db_url)
def shuffle_random_func(self):
return self.shuffle_random_seed
@ -1165,103 +1170,36 @@ class CLITestLogger(TestLogger):
return log_line_str
class BaseDBAccess():
""" Class used to connect with test database and store test results
"""
def __init__(self):
self.db_object = None
# Test Suite DB scheme (table names)
self.TABLE_BUILD_ID = 'mtest_build_id'
self.TABLE_BUILD_ID_STATUS = 'mtest_build_id_status'
self.TABLE_TARGET = 'mtest_target'
self.TABLE_TEST_ENTRY = 'mtest_test_entry'
self.TABLE_TEST_ID = 'mtest_test_id'
self.TABLE_TEST_RESULT = 'mtest_test_result'
self.TABLE_TEST_TYPE = 'mtest_test_type'
self.TABLE_TOOLCHAIN = 'mtest_toolchain'
# Build ID status PKs
self.BUILD_ID_STATUS_STARTED = 1 # Started
self.BUILD_ID_STATUS_IN_PROGRESS = 2 # In Progress
self.BUILD_ID_STATUS_COMPLETED = 3 #Completed
self.BUILD_ID_STATUS_FAILED = 4 # Failed
def factory_db_logger(db_url):
(db_type, username, password, host, db_name) = BaseDBAccess().parse_db_connection_string(db_url)
if db_type == 'mysql':
return MySQLDBAccess()
else:
return None
def get_hostname(self):
""" Useful when creating build_id in database
Function returns (hostname, uname) which can be used as (build_id_name, build_id_desc)
"""
# Get hostname from socket
import socket
hostname = socket.gethostbyaddr(socket.gethostname())[0]
# Get uname from platform resources
import platform
uname = json.dumps(platform.uname())
return (hostname, uname)
def parse_db_connection_string(self, str):
""" Parsing SQL DB connection string. String should contain:
- DB Name, user name, password, URL (DB host), name
Function should return tuple with parsed (host, user, passwd, db) or None if error
E.g. connection string: 'mysql://username:password@127.0.0.1/db_name'
"""
PATTERN = '^([\w]+)://([\w]+):([\w]*)@(.*)/([\w]+)'
result = re.match(PATTERN, str)
if result is not None:
result = result.groups() # Tuple (db_name, host, user, passwd, db)
return result
def is_connected(self):
""" Returns True if we are connected to database
"""
pass
def connect(self, host, user, passwd, db):
""" Connects to DB and returns DB object
"""
pass
def disconnect(self):
""" Close DB connection
"""
pass
def escape_string(self, str):
""" Escapes string so it can be put in SQL query between quotes
"""
pass
def select_all(self, query):
""" Execute SELECT query and get all results
"""
pass
def insert(self, query, commit=True):
""" Execute INSERT query, define if you want to commit
"""
pass
def get_next_build_id(self, name, desc=''):
""" Insert new build_id (DB unique build like ID number to send all test results)
"""
pass
def get_table_entry_pk(self, table, column, value, update_db=True):
""" Checks for entries in tables with two columns (<TABLE_NAME>_pk, <column>)
If update_db is True updates table entry if value in specified column doesn't exist
"""
pass
def update_table_entry(self, table, column, value):
""" Updates table entry if value in specified column doesn't exist
Locks table to perform atomic read + update
"""
pass
def insert_test_entry(self, next_build_id, target, toolchain, test_type, test_id, test_result, test_time, test_timeout, test_loop, test_extra=''):
""" Inserts test result entry to database. All checks regarding existing
toolchain names in DB are performed.
If some data is missing DB will be updated
"""
pass
def detect_database_verbose(db_url):
result = BaseDBAccess().parse_db_connection_string(db_url)
if result is not None:
# Parsing passed
(db_type, username, password, host, db_name) = result
#print "DB type '%s', user name '%s', password '%s', host '%s', db name '%s'"% result
# Let's try to connect
db_ = factory_db_logger(db_url)
if db_ is not None:
print "Connecting to database '%s'..."% db_url,
db_.connect(host, username, password, db_name)
if db_.is_connected():
print "ok"
print "Detecting database..."
print db_.detect_database(verbose=True)
print "Disconnecting...",
db_.disconnect()
print "done"
else:
print "Database type '%s' unknown"% db_type
else:
print "Parse error: '%s' - DB Url error"% (db_url)
def get_default_test_options_parser():
@ -1397,6 +1335,10 @@ def get_default_test_options_parser():
type="int",
help='You can increase global timeout for each test by specifying additional test timeout in seconds')
parser.add_option('', '--db',
dest='db_url',
help='This specifies what database test suite uses to store its state. To pass DB connection info use database connection string. Example: \'mysql://username:password@127.0.0.1/db_name\'')
parser.add_option('-l', '--log',
dest='log_file_name',
help='Log events to external file (note not all console entries may be visible in log file)')

130
workspace_tools/test_db.py Normal file
View File

@ -0,0 +1,130 @@
"""
mbed SDK
Copyright (c) 2011-2014 ARM Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
"""
import re
class BaseDBAccess():
""" Class used to connect with test database and store test results
"""
def __init__(self):
self.db_object = None
self.db_type = None
# Test Suite DB scheme (table names)
self.TABLE_BUILD_ID = 'mtest_build_id'
self.TABLE_BUILD_ID_STATUS = 'mtest_build_id_status'
self.TABLE_TARGET = 'mtest_target'
self.TABLE_TEST_ENTRY = 'mtest_test_entry'
self.TABLE_TEST_ID = 'mtest_test_id'
self.TABLE_TEST_RESULT = 'mtest_test_result'
self.TABLE_TEST_TYPE = 'mtest_test_type'
self.TABLE_TOOLCHAIN = 'mtest_toolchain'
# Build ID status PKs
self.BUILD_ID_STATUS_STARTED = 1 # Started
self.BUILD_ID_STATUS_IN_PROGRESS = 2 # In Progress
self.BUILD_ID_STATUS_COMPLETED = 3 #Completed
self.BUILD_ID_STATUS_FAILED = 4 # Failed
def get_hostname(self):
""" Useful when creating build_id in database
Function returns (hostname, uname) which can be used as (build_id_name, build_id_desc)
"""
# Get hostname from socket
import socket
hostname = socket.gethostbyaddr(socket.gethostname())[0]
# Get uname from platform resources
import platform
uname = json.dumps(platform.uname())
return (hostname, uname)
def get_db_type(self):
""" Returns database type. E.g. 'mysql', 'sqlLite' etc.
"""
return self.db_type
def detect_database(self, verbose=False):
""" detect database and return VERION data structure or string (verbose=True)
"""
return None
def parse_db_connection_string(self, str):
""" Parsing SQL DB connection string. String should contain:
- DB Name, user name, password, URL (DB host), name
Function should return tuple with parsed (db_type, username, password, host, db_name) or None if error
E.g. connection string: 'mysql://username:password@127.0.0.1/db_name'
"""
PATTERN = '^([\w]+)://([\w]+):([\w]*)@(.*)/([\w]+)'
result = re.match(PATTERN, str)
if result is not None:
result = result.groups() # Tuple (db_name, host, user, passwd, db)
return result # (db_type, username, password, host, db_name)
def is_connected(self):
""" Returns True if we are connected to database
"""
pass
def connect(self, host, user, passwd, db):
""" Connects to DB and returns DB object
"""
pass
def disconnect(self):
""" Close DB connection
"""
pass
def escape_string(self, str):
""" Escapes string so it can be put in SQL query between quotes
"""
pass
def select_all(self, query):
""" Execute SELECT query and get all results
"""
pass
def insert(self, query, commit=True):
""" Execute INSERT query, define if you want to commit
"""
pass
def get_next_build_id(self, name, desc=''):
""" Insert new build_id (DB unique build like ID number to send all test results)
"""
pass
def get_table_entry_pk(self, table, column, value, update_db=True):
""" Checks for entries in tables with two columns (<TABLE_NAME>_pk, <column>)
If update_db is True updates table entry if value in specified column doesn't exist
"""
pass
def update_table_entry(self, table, column, value):
""" Updates table entry if value in specified column doesn't exist
Locks table to perform atomic read + update
"""
pass
def insert_test_entry(self, next_build_id, target, toolchain, test_type, test_id, test_result, test_time, test_timeout, test_loop, test_extra=''):
""" Inserts test result entry to database. All checks regarding existing
toolchain names in DB are performed.
If some data is missing DB will be updated
"""
pass

27
workspace_tools/test_mysql.py
View File

@ -20,9 +20,8 @@ Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
import re
import MySQLdb as mdb
# Imports from TEST API
from workspace_tools.test_api import BaseDBAccess
from workspace_tools.test_db import BaseDBAccess
class MySQLDBAccess(BaseDBAccess):
@ -31,6 +30,20 @@ class MySQLDBAccess(BaseDBAccess):
def __init__(self):
BaseDBAccess.__init__(self)
def detect_database(self, verbose=False):
""" detect database and return VERION data structure or string (verbose=True)
"""
query = 'SHOW VARIABLES LIKE "%version%"'
rows = self.select_all(query)
if verbose:
result = []
for row in rows:
result.append("\t%s: %s"% (row['Variable_name'], row['Value']))
result = "\n".join(result)
else:
result = rows
return result
def parse_db_connection_string(self, str):
""" Parsing SQL DB connection string. String should contain:
- DB Name, user name, password, URL (DB host), name
@ -38,8 +51,10 @@ class MySQLDBAccess(BaseDBAccess):
E.g. connection string: 'mysql://username:password@127.0.0.1/db_name'
"""
result = BaseDBAccess.parse_db_connection_string(str)
if result is not None and result[0] != 'mysql':
result = None
if result is not None:
(db_type, username, password, host, db_name) = result
if db_type != 'mysql':
result = None
return result
def is_connected(self):
@ -52,15 +67,19 @@ class MySQLDBAccess(BaseDBAccess):
"""
try:
self.db_object = mdb.connect(host=host, user=user, passwd=passwd, db=db)
self.db_type = 'mysql'
except mdb.Error, e:
print "Error %d: %s"% (e.args[0], e.args[1])
self.db_object = None
self.db_type = None
def disconnect(self):
""" Close DB connection
"""
if self.db_object:
self.db_object.close()
self.db_object = None
self.db_type = None
def escape_string(self, str):
""" Escapes string so it can be put in SQL query between quotes