KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use tb_info method in avocado
Best Python code snippet using avocado_python
fetch_tb_info_db.py
Source:fetch_tb_info_db.py 
1__author__ = 'UKumar'2import sys3import re4from string import ascii_uppercase5from dbHandler import dbHandler6from fetch_testbed_info import FetchTestbedInfo7class FetchTbInfoDatabase:8 """9 classdocs10 """11 def __init__(self, input, logger):12 """13 Constructor14 """15 self.input = input16 self.logger = logger17 self.logger.debug('Begin')18 self.dbObj = dbHandler()19 self.logger.debug('End')20 def fetch_mh_st_info(self):21 tb_info = dict()22 #tb_info_db = self.select(tb_name)23 pid = self.dbObj.get_project_id(self.input['project'])24 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"25 # query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project='"+self.input['project']+"' and type='"+self.input['testbed']['type']+"'"26 print "query1 :", query127 tb_id = self.dbObj.execute_select_query(query1)28 print "id:", tb_id[0][0]29 print "id is ", int(str(tb_id).lstrip("(").rstrip("L,)"))30 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE name=\'HQ\' and testbed_id='+str(tb_id[0][0])31 print hqQuery32 hqInfo = self.dbObj.execute_select_query(hqQuery)33 print "hq info: ", hqInfo34 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])35 tb_info['SITE1_USERNAME'] = str(hqInfo[0][1])36 tb_info['SITE1_PASSWORD'] = str(hqInfo[0][2])37 tb_info['OS_USERNAME'] = str(hqInfo[0][3])38 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])39 phoneQuery = 'SELECT mac FROM rodexecute_device WHERE name=\'IPPhone\' and testbed_id='+str(tb_id[0][0])40 phonemac = self.dbObj.execute_select_query(phoneQuery)41 print phonemac42 i = 143 for mac in phonemac:44 tb_info['PHONE'+str(i)+'_MAC_ADDR'] = mac[0]45 i = i+146 print tb_info47 return tb_info48 def fetch_mh_mt_info(self, tb_info):49 # query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project='"+self.input['project']+"' and type='"+self.input['testbed']['type']+"'"50 pid = self.dbObj.get_project_id(self.input['project'])51 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"52 print "query1 :", query153 tb_id = self.dbObj.execute_select_query(query1)54 print "id:", tb_id[0][0]55 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE name=\'HQ\' and testbed_id='+str(tb_id[0][0])56 print hqQuery57 hqInfo = self.dbObj.execute_select_query(hqQuery)58 print "hq info: ", hqInfo59 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])60 tb_info['MT_D2_SERVER_IP'] = str(hqInfo[0][0])61 tb_info['MT_D2_LOGINNAME'] = str(hqInfo[0][1])62 tb_info['MT_D2_LOGINPASSWORD'] = str(hqInfo[0][2])63 tb_info['OS_USERNAME'] = str(hqInfo[0][3])64 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])65 print tb_info66 switchQuery = 'SELECT name, ip, type, mac FROM rodexecute_device WHERE type=\'SG-vPhone\' and testbed_id='+str(tb_id[0][0])67 switches = self.dbObj.execute_select_query(switchQuery)68 print switches69 i = 170 for switch in switches:71 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_NAME'] = switch[0]72 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_ADDR'] = switch[1]73 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_TYPE'] = switch[2]74 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_MAC'] = switch[3]75 i = i+176 tenantQuery = 'SELECT name, tenant_id, prefix FROM rodexecute_tenants WHERE testbed_id='+str(tb_id[0][0])77 tenants = self.dbObj.execute_select_query(tenantQuery)78 print "tenant: ", tenants79 tb_info['MT_DEFAULT_TENANT_NAME'] = tenants[0][0]80 tb_info['MT_DEFAULT_TENANT_ID'] = tenants[0][1]81 tb_info['MT_DEFAULT_TENANT_PREFIX'] = tenants[0][2]82 testInfo = self.dbObj.execute_select_query('SELECT * FROM rodexecute_manhattantestdata WHERE testbed_id='+str(tb_id[0][0]))83 tb_info['MT_ABC_AUTHENTICATION_URL'] = testInfo[0][1]84 tb_info['REVERSE_PROXY_IP_FOR_HOST'] = testInfo[0][2]85 #print tb_info86 return tb_info87 def fetch_mb_st_info(self, tb_info):88 usrmac_info = dict()89 pid = self.dbObj.get_project_id(self.input['project'])90 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"91 # query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project='"+self.input['project']+"' and type='"+self.input['testbed']['type']+"'"92 print "query1 :", query193 tb_id = self.dbObj.execute_select_query(query1)94 print "id:", tb_id[0][0]95 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE (name=\'HQ\' OR name=\'SMR\') and testbed_id='+str(tb_id[0][0])96 print hqQuery97 hqInfo = self.dbObj.execute_select_query(hqQuery)98 print "hq info: ", hqInfo99 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])100 tb_info['SITE1_USERNAME'] = str(hqInfo[0][1])101 tb_info['SITE1_PASSWORD'] = str(hqInfo[0][2])102 tb_info['OS_USERNAME'] = str(hqInfo[0][3])103 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])104 tb_info['SMR_IP_ADDR'] = str(hqInfo[1][0])105 tb_info['SMR_USERNAME'] = str(hqInfo[1][1])106 tb_info['SMR_PASSWORD'] = str(hqInfo[1][2])107 mDeviceQuery = 'SELECT port, plateform_name, plateform_version, name, udid, ios_udid, password ' \108 'FROM rodexecute_mobilitydevices WHERE testbed_id='+str(tb_id[0][0])109 mobilityDevices = self.dbObj.execute_select_query(mDeviceQuery)110 print mobilityDevices111 i = 1112 for device in mobilityDevices:113 tb_info['DEVICE'+str(i)+'_PORT'] = device[0]114 tb_info['DEVICE'+str(i)+'_PLATFORM_NAME'] = device[1]115 tb_info['DEVICE'+str(i)+'_PLATFORM_VERSION'] = device[2]116 tb_info['DEVICE'+str(i)+'_NAME'] = device[3]117 tb_info['DEVICE'+str(i)+'_UDID'] = device[4]118 tb_info['DEVICE'+str(i)+'_PASSWORD'] = device[6]119 tb_info['DEVICE1_iOS_UDID'] = mobilityDevices[0][5]120 mbTestDataQuery = 'SELECT * FROM rodexecute_mobilitytestdata WHERE testbed_id='+str(tb_id[0][0])121 mbTestData = self.dbObj.execute_select_query(mbTestDataQuery)122 tb_info['APP_PKG'] = mbTestData[0][1]123 tb_info['APP_ACTIVITY'] = mbTestData[0][2]124 tb_info['APP_LOCATION'] = mbTestData[0][3]125 tb_info['NEW_COMMAND_TIMEOUT'] = mbTestData[0][4]126 phoneType = self.dbObj.execute_select_query('SELECT type FROM rodexecute_device WHERE name=\'IPPhone\' and testbed_id='+str(tb_id[0][0]))[0][0]127 mbUsersQuery = 'SELECT * FROM rodexecute_usernames WHERE testbed_id='+str(tb_id[0][0])128 mbUsers = self.dbObj.execute_select_query(mbUsersQuery)129 print mbUsers130 i = 1131 noOfUsers = len(mbUsers)132 for ch in ascii_uppercase:133 tb_info['USER_'+ch+'_NAME'] = mbUsers[i-1][1]134 tb_info['USER_'+ch+'_FULLNAME'] = mbUsers[i-1][2]+"#"+mbUsers[i-1][3]135 tb_info['USER_'+ch+'_EXTENSION'] = mbUsers[i-1][4]136 tb_info['PPHONE_0'+str(i)+'_ADDR'] = str(mbUsers[i-1][6]) # for converting None to string None137 usrmac_info['USER'+str(i)+'_FIRSTNAME'] = mbUsers[i-1][2]138 usrmac_info['USER'+str(i)+'_LASTNAME'] = mbUsers[i-1][3]139 usrmac_info['USER'+str(i)+'_EXTENSION'] = mbUsers[i-1][4]140 usrmac_info['USER'+str(i)+'_DID'] = str(mbUsers[i-1][5])141 #usrmac_info['PPHONE_0'+str(i)+'_ADDR'] = str(mbUsers[i-1][7])142 usrmac_info['USER'+str(i)+'_MAC'] = re.sub('\-', '', str(mbUsers[i-1][7]))143 if i == noOfUsers:144 break145 i += 1146 usrmac_info['NUMBER_OF_USERS'] = len(mbUsers)147 usrmac_info['PHONE_TYPE'] = str(phoneType)148 #print tb_info149 #print usrmac_info150 return tb_info, usrmac_info151 def fetch_pbx_mt_info(self, tb_info):152 usrmac_info = dict()153 bco_config_info = dict()154 pid = self.dbObj.get_project_id(self.input['project'])155 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"156 # query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project='"+self.input['project']+"' and type='"+self.input['testbed']['type']+"'"157 print "query1 :", query1158 tb_id = self.dbObj.execute_select_query(query1)159 print "id:", tb_id160 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE name=\'HQ\' and testbed_id='+str(tb_id[0][0])161 print hqQuery162 hqInfo = self.dbObj.execute_select_query(hqQuery)163 print "hq info: ", hqInfo164 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])165 tb_info['MT_D2_SERVER_IP'] = str(hqInfo[0][0])166 tb_info['MT_D2_LOGINNAME'] = str(hqInfo[0][1])167 tb_info['MT_D2_LOGINPASSWORD'] = str(hqInfo[0][2])168 tb_info['OS_USERNAME'] = str(hqInfo[0][3])169 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])170 #print tb_info171 switchQuery = 'SELECT name, ip, type, mac FROM rodexecute_device WHERE type=\'SG-vPhone\' and testbed_id='+str(tb_id[0][0])172 switches = self.dbObj.execute_select_query(switchQuery)173 i = 1174 for switch in switches:175 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_NAME'] = switch[0]176 bco_config_info['SITE1_SWITCH'+str(i)+'_NAME'] = switch[0]177 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_ADDR'] = switch[1]178 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_TYPE'] = switch[2]179 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_MAC'] = switch[3]180 i = i+1181 tenantQuery = 'SELECT name, tenant_id, prefix FROM rodexecute_tenants WHERE testbed_id='+str(tb_id[0][0])182 tenants = self.dbObj.execute_select_query(tenantQuery)183 tb_info['MT_DEFAULT_TENANT_NAME'] = tenants[0][0]184 tb_info['MT_DEFAULT_TENANT_ID'] = tenants[0][1]185 tb_info['MT_DEFAULT_TENANT_PREFIX'] = tenants[0][2]186 testInfo = self.dbObj.execute_select_query('SELECT * FROM rodexecute_pbxtestdata WHERE testbed_id='+str(tb_id[0][0]))187 tb_info['MT_ABC_AUTHENTICATION_URL'] = testInfo[0][1]188 tb_info['REVERSE_PROXY_IP_FOR_HOST'] = testInfo[0][2]189 bco_config_info['PRI_DID_RANGE'] = testInfo[0][3]190 pbxUsersQuery = 'SELECT * FROM rodexecute_usernames WHERE testbed_id='+str(tb_id[0][0])191 pbxUsers = self.dbObj.execute_select_query(pbxUsersQuery)192 noOfUsers = len(pbxUsers)193 for i in range(1, noOfUsers+1):194 tb_info['USER'+str(i)+'_NAME'] = pbxUsers[i-1][2]195 tb_info['PPHONE_0'+str(i)+'_ADDR'] = str(pbxUsers[i-1][6]) # for converting None to string None196 bco_config_info['PPHONE_0'+str(i)+'_ADDR'] = str(pbxUsers[i-1][6])197 usrmac_info['USER'+str(i)+'_FIRSTNAME'] = pbxUsers[i-1][2]198 usrmac_info['USER'+str(i)+'_LASTNAME'] = pbxUsers[i-1][3]199 usrmac_info['USER'+str(i)+'_EXTENSION'] = pbxUsers[i-1][4]200 usrmac_info['USER'+str(i)+'_DID'] = str(pbxUsers[i-1][5])201 usrmac_info['USER'+str(i)+'_MAC'] = re.sub('\-', '', str(pbxUsers[i-1][7]))202 # if i == noOfUsers:203 # break204 # i += 1205 usrmac_info['NUMBER_OF_USERS'] = len(pbxUsers)206 phoneType = self.dbObj.execute_select_query('SELECT type FROM rodexecute_device WHERE name=\'IPPhone\' and testbed_id='+str(tb_id[0][0]))[0][0]207 usrmac_info['PHONE_TYPE'] = str(phoneType)208 ldvsswitchQuery = 'SELECT name FROM rodexecute_device WHERE type=\'SG-LinuxDVS\' and testbed_id='+str(tb_id[0][0])209 ldvsSwitches = self.dbObj.execute_select_query(ldvsswitchQuery)210 i = 1211 for ldvsSwitch in ldvsSwitches:212 bco_config_info['BCO_LINUXDVS_SWITCH'+str(i)+'_NAME'] = ldvsSwitch[0]213 i = i+1214 bco_config_info['HQ_LOGIN_NAME'] = tb_info['MT_D2_LOGINNAME']215 bco_config_info['MT_HQ_SERVER'] = tb_info['MT_D2_LOGINNAME']216 bco_config_info['REVERSE_PROXY_IP'] = tb_info['MT_ABC_AUTHENTICATION_URL']217 bco_config_info['BCO_PPHONE_SITE1_SERVER_ADDR'] = tb_info['MT_D2_SERVER_IP']218 #print tb_info219 return tb_info, usrmac_info, bco_config_info220 def fetch_mnm_st_info(self):221 tb_info = dict()222 usrmac_info = dict()223 #tb_info_db = self.select(tb_name)224 pid = self.dbObj.get_project_id(self.input['project'])225 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"226 print "query1 :", query1227 tb_id = self.dbObj.execute_select_query(query1)228 print "id:", tb_id[0][0]229 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE name=\'HQ\' and testbed_id='+str(tb_id[0][0])230 print hqQuery231 hqInfo = self.dbObj.execute_select_query(hqQuery)232 print "hq info: ", hqInfo233 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])234 tb_info['SITE1_USERNAME'] = str(hqInfo[0][1])235 tb_info['SITE1_PASSWORD'] = str(hqInfo[0][2])236 tb_info['OS_USERNAME'] = str(hqInfo[0][3])237 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])238 ''' fetch user info and add to dict'''239 mnmUsersQuery = 'SELECT * FROM rodexecute_usernames WHERE testbed_id='+str(tb_id[0][0])240 mnmUsers = self.dbObj.execute_select_query(mnmUsersQuery)241 print mnmUsers242 noOfUsers = len(mnmUsers)243 for i in range(1, noOfUsers+1):244 tb_info['USER'+str(i)+'_NAME'] = mnmUsers[i-1][2]245 tb_info['PPHONE_0'+str(i)+'_ADDR'] = str(mnmUsers[i-1][6]) # for converting None to string None246 usrmac_info['USER'+str(i)+'_FIRSTNAME'] = mnmUsers[i-1][2]247 usrmac_info['USER'+str(i)+'_LASTNAME'] = mnmUsers[i-1][3]248 usrmac_info['USER'+str(i)+'_EXTENSION'] = mnmUsers[i-1][4]249 usrmac_info['USER'+str(i)+'_DID'] = str(mnmUsers[i-1][5])250 usrmac_info['USER'+str(i)+'_MAC'] = re.sub('\-', '', str(mnmUsers[i-1][7]))251 usrmac_info['NUMBER_OF_USERS'] = len(mnmUsers)252 phoneType = self.dbObj.execute_select_query('SELECT type FROM rodexecute_device WHERE name=\'IPPhone\' and testbed_id='+str(tb_id[0][0]))[0][0]253 usrmac_info['PHONE_TYPE'] = str(phoneType)254 '''fetch testdata and insert into dict'''255 mnmTestDataQuery = 'SELECT * FROM rodexecute_mnmtestdata WHERE testbed_id='+str(tb_id[0][0])256 #print "data query: ", mnmTestDataQuery257 mnmTestData = self.dbObj.execute_select_query(mnmTestDataQuery)258 print mnmTestData259 for i in range(len(mnmTestData)):260 c = i+1261 tb_info['Off_System_Extensions'+str(c)] = mnmTestData[i-1][1]262 print tb_info['Off_System_Extensions'+str(c)]263 ''' fetch switch info and insert into tb_info'''264 switchQuery = 'SELECT name, ip, type, mac FROM rodexecute_device WHERE type=\'SG-vPhone\' and testbed_id='+str(tb_id[0][0])265 print switchQuery266 switches = self.dbObj.execute_select_query(switchQuery)267 print switches268 i = 1269 for switch in switches:270 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_NAME'] = switch[0]271 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_ADDR'] = switch[1]272 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_TYPE'] = switch[2]273 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_MAC'] = switch[3]274 i = i+1275 #print "TB INFO IS : ", tb_info276 #print "USERMAC INFO ", usrmac_info277 return tb_info , usrmac_info278 def fetch_mnm_mt_info(self, tb_info):279 '''280 :return: will return mnm project test info from mysqql db281 '''282 usrmac_info = dict()283 #tb_info_db = self.select(tb_name)284 pid = self.dbObj.get_project_id(self.input['project'])285 query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project_id='"+str(pid)+"' and type='"+self.input['testbed']['type']+"'"286 # query1 = 'SELECT id FROM rodexecute_testbed WHERE name='+"'"+self.input['testbed_name']+"' and ownedby='"+self.input['user']+"' and project='"+self.input['project']+"' and type='"+self.input['testbed']['type']+"'"287 print "query1 :", query1288 tb_id = self.dbObj.execute_select_query(query1)289 print "id:", tb_id[0][0]290 hqQuery = 'SELECT ip, app_user, app_password, os_user, os_password FROM rodexecute_device WHERE name=\'HQ\' and testbed_id='+str(tb_id[0][0])291 print hqQuery292 hqInfo = self.dbObj.execute_select_query(hqQuery)293 print "hq info: ", hqInfo294 tb_info['SITE1_SERVER_ADDR'] = str(hqInfo[0][0])295 tb_info['MT_D2_SERVER_IP'] = str(hqInfo[0][0])296 tb_info['MT_D2_LOGINNAME'] = str(hqInfo[0][1])297 tb_info['MT_D2_LOGINPASSWORD'] = str(hqInfo[0][2])298 tb_info['OS_USERNAME'] = str(hqInfo[0][3])299 tb_info['OS_PASSWORD'] = str(hqInfo[0][4])300 ''' fetch user info and add to dict'''301 mnmUsersQuery = 'SELECT * FROM rodexecute_usernames WHERE testbed_id='+str(tb_id[0][0])302 mnmUsers = self.dbObj.execute_select_query(mnmUsersQuery)303 print mnmUsers304 noOfUsers = len(mnmUsers)305 for i in range(1, noOfUsers+1):306 tb_info['USER'+str(i)+'_NAME'] = mnmUsers[i-1][2]307 tb_info['PPHONE_0'+str(i)+'_ADDR'] = str(mnmUsers[i-1][6]) # for converting None to string None308 usrmac_info['USER'+str(i)+'_FIRSTNAME'] = mnmUsers[i-1][2]309 usrmac_info['USER'+str(i)+'_LASTNAME'] = mnmUsers[i-1][3]310 usrmac_info['USER'+str(i)+'_EXTENSION'] = mnmUsers[i-1][4]311 usrmac_info['USER'+str(i)+'_DID'] = str(mnmUsers[i-1][5])312 usrmac_info['USER'+str(i)+'_MAC'] = re.sub('\-', '', str(mnmUsers[i-1][7]))313 usrmac_info['NUMBER_OF_USERS'] = len(mnmUsers)314 phoneType = self.dbObj.execute_select_query('SELECT type FROM rodexecute_device WHERE name=\'IPPhone\' and testbed_id='+str(tb_id[0][0]))[0][0]315 usrmac_info['PHONE_TYPE'] = str(phoneType)316 '''fetch testdata and insert into dict'''317 mnmTestDataQuery = 'SELECT * FROM rodexecute_mnmtestdata WHERE testbed_id='+str(tb_id[0][0])318 print "data query: ", mnmTestDataQuery319 mnmTestData = self.dbObj.execute_select_query(mnmTestDataQuery)320 #print mnmTestData321 tb_info['MT_ABC_AUTHENTICATION_URL'] = mnmTestData[0][2]322 tb_info['REVERSE_PROXY_IP_FOR_HOST'] = mnmTestData[0][3]323 ''' fetch switch info and insert into tb_info'''324 switchQuery = 'SELECT name, ip, type, mac FROM rodexecute_device WHERE type=\'SG-vPhone\' and testbed_id='+str(tb_id[0][0])325 print switchQuery326 switches = self.dbObj.execute_select_query(switchQuery)327 print switches328 i = 1329 for switch in switches:330 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_NAME'] = switch[0]331 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_ADDR'] = switch[1]332 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_TYPE'] = switch[2]333 tb_info['SETUP'+str(i)+'_SITE'+str(i)+'_SWITCH'+str(i)+'_MAC'] = switch[3]334 i = i+1335 ''' fetch tenant info and insert into tb_info dict'''336 tenantQuery = 'SELECT name, tenant_id, prefix FROM rodexecute_tenants WHERE testbed_id='+str(tb_id[0][0])337 tenants = self.dbObj.execute_select_query(tenantQuery)338 tb_info['MT_DEFAULT_TENANT_NAME'] = tenants[0][0]339 tb_info['MT_DEFAULT_TENANT_ID'] = tenants[0][1]340 tb_info['MT_DEFAULT_TENANT_PREFIX'] = tenants[0][2]341 print tb_info, usrmac_info342 return tb_info, usrmac_info343 def select(self, tbname):344 row = (1, u'mh-st_10.198.166.44', u'10.198.166.44', u'admin', u'changeme', u'0.0.0.0,1.1.1.1,2.2.2.2')345 return row346if __name__ == '__main__':347 ftb = FetchTbInfoDatabase()...
event.py
Source:event.py
1import time2from datetime import datetime3from const import CheckResult4from tb_info import TracebackInfo5class Event:6 def __init__(self): # noqa7 self.timestamp = time.time()8 def get_utc_timestamp(self) -> datetime:9 return datetime.utcfromtimestamp(self.timestamp)10 def get_timestamp(self) -> datetime:11 """Returns timestamp for local tz."""12 return datetime.fromtimestamp(self.timestamp)13class FailEvent(Event):14 def __init__(self, message=''):15 super().__init__()16 self.message = message17 def __repr__(self):18 # no timestamp19 return f'FailEvent(message="{self.message}")'20class ErrorEvent(Event):21 def __init__(self, tb_info: TracebackInfo = None):22 super().__init__()23 self.tb_info = tb_info24 def __repr__(self):25 # no timestamp26 if self.tb_info is None:27 tb_info = None28 else:29 tb_info = f'TracebackInfo({self.tb_info})'30 return f'ErrorEvent(tb_info={tb_info})'31class CheckEvent(Event):32 def __init__(self, result: CheckResult = None, tb_info: TracebackInfo = None, message=''):33 super().__init__()34 self.result = result35 self.message = message36 self.tb_info = tb_info37class Check2Event(CheckEvent):38 def __init__(self, *args, actual=None, sign=None, expected=None, strict=None, **kwargs):39 super().__init__(*args, **kwargs)40 self.actual = actual41 self.sign = sign42 self.expected = expected43 self.strict = strict44 def __repr__(self):45 # no timestamp46 if isinstance(self.actual, str):47 actual = f'"{self.actual}"'48 else:49 actual = self.actual50 if isinstance(self.expected, str):51 expected = f'"{self.expected}"'52 else:53 expected = self.expected54 if self.tb_info is None:55 tb_info = None56 else:57 tb_info = f'TracebackInfo({self.tb_info})'58 return (59 f'Check2Event('60 f'message="{self.message}"'61 f', actual={actual}'62 f', sign="{self.sign}"'63 f', expected={expected}'64 f', result=CheckResult.{self.result.name}'65 f', strict={self.strict}'66 f', tb_info={tb_info}'67 f')'...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run avocado automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
