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')'...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
