Multithreading realizes concurrent reading and execution of multi use case files +selenium grid4 realizes distributed deployment of test framework

---

1.main.py

import os
import threading

from class25.excel_driver.excel_read import path, read
from class26 import log_conf

if __name__ == '__main__':
    #  establish log object 
    log = log_conf.get_log('./conf/log.ini')
    #  By reading the data Whether there are test cases under the path , Yes , Is executed , No, No 
    #  Rerun set 
    repeat = list()
    #  Test case collection 
    cases = list()
    #  Thread pool creation 
    th = list()

    #  Read the test cases under the path 
    for path,dir,files in os.walk('../class25/data/'):
        for file in files:
            #  Get the suffix of the file 
            file_type = os.path.splitext(file)[1]
            # file_name = os.path.split(file)[1]
            file_name = os.path.splitext(file)[0]
            # print(file_type)
            # print(path+file)
            if file_type == '.xlsx':
                # print(file_type)
                if 'old' not in file_name:
                    case_path = path+file
                    cases.append(case_path)
            else:
                log.error(' Wrong file type :{}'.format(file))
    # for case in cases:
    # log.info('******** Being implemented {} file *********'.format(case))
    # f = read(case,log)
    # if f:
    # repeat.append(f)

    # if len(repeat)>0:
    # # Call rerun , function repeat All use cases under 
    # pass

    #  Set the corresponding number of threads based on the number of use cases 
    for case in cases:
        th.append(threading.Thread(target = read,args = [case,log]))

    #  Start thread 
    for t in th:
        t.start()

2. Encapsulate the suite function that gets the use case

'''  Encapsulate the suite function that gets the use case  '''
import threading
import unittest


def get_suite():
    path = './'
    discover = unittest.defaultTestLoader.discover(start_dir=path, pattern='test*.py')
    return discover


#  The runner runs the suite functions 
def runner(suite):
    unittest.TextTestRunner().run(suite)

if __name__ == '__main__':
    suites = get_suite()
    th = []
    for suite in suites:
        th.append(threading.Thread(target=runner,args = [suite]))
    for t in th:
        t.start()

3. Multithreaded use case concurrent processing implementation

'''  Multithreaded use case concurrent processing implementation   Multithreaded operation  '''
import threading
from time import sleep


def run(name):
    print(name+ ' Running ')
    sleep(3)
    print(name+' Tired of running ')

1. Single thread effect

li = [' fox ',' cyber-love experiences ',' Cold lamp ']
# for i in li:
# run(i)

2. Set up thread pool

#  Run together : Realize the effect of multi-user running together through multi-threaded form 
th = list()

3. Assign threads

th.append(threading.Thread(target = run,args=[' fox ']))
th.append(threading.Thread(target = run,args=[' cyber-love experiences ']))
th.append(threading.Thread(target = run,args=[' Cold lamp ']))

4. Start thread

#  If your multiple threads are reading and writing to the same file , To prevent thread lock , Need to add join
for t in th:
    t.start()
    #  In order to avoid multiple threads operating on the same file or content , Lock caused by 
    t.join()

4. visit Selenium Grid system

'''  visit Selenium Grid system  '''
from time import sleep

from selenium import webdriver
#  establish remote object 

driver = webdriver.remote('http://127.0.0.1:4444/wd/hub')
# a Business ,b Business ,c Business , Using different driver Objects can perform different businesses 
driver.get('http://www.baidu.com')
sleep(5)
driver.quit()