总体架构组成

logger.py 顾名思义，配置logger的

import logginglogger = logging.getLogger('Base_logger')
logger.setLevel(logging.DEBUG)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(filename)s[:%(lineno)d] - %(message)s"))
stream_handler.setLevel(logging.DEBUG)
logger.addHandler(stream_handler)base_logger = logging.getLogger('Tester_logger')
base_logger.setLevel(logging.DEBUG)
base_stream = logging.StreamHandler()
base_stream.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(function_name)s- %(error)s - %(message)s"))
base_stream.setLevel(logging.DEBUG)
base_logger.addHandler(base_stream)

base.py 一些基础的方法

• ExcelHandler：Excel文件处理类
• ***toGet()，toPost()，toSeriesRequest()***：三个分别对httpx的get(), post()，以及流程化请求方法的一些封装。增加对超时异常的重跑功能(高并发会出现较多的超时异常，可以通过配置timeout，或利用异常重跑)。后期扩展，可以对其他不同的请求方法进行扩展。
• TestBase： 测试运行的基类，目前主要用来获取配置内容–config，读取用例excel文件后获取的dataframe，配置client。
  余下类方法的均用作使用说明。 test_cases为日志文件中每一页sheet的dataframe(pandas)，可以通过增加不同的sheet，并增加test_cases，以适配不同类型的接口。

class TestBase:config = load()test_cases = ExcelHandler.get_testcases()  # config diff sheet_name params to get dataframes from sheetsdef __init__(self):self.client = httpx.AsyncClient(verify=False)async def fetch(self, semaphore):'''A function to rewrite for make a request!!Example:async with semaphore:resp = await self.client.get(url)'''pass

tester.py 配置各个接口

• 一般把参数作为变量传入，被调用时，利用用例文件中的内容对其传参。同时对获取到的响应进行处理，这部分自由发挥，这里把结果，提取出请求参数，又写进excel用例文件中的<测试结果列>。

class Test(base.TestBase):def __init__(self):super().__init__()async def fetch_1(self, semaphore, data, test_id):async with semaphore:resp = await base.toPost(self.client, url='http://httpbin.org/post', data=data, timeout=(30.07, 27))if resp:ret = resp.json()['data']await ExcelHandler.write_response(self.test_cases, test_id, ret)

TestPlan.py 读取用例传入配置的接口，加载到事件循环中，并执行
事件循环是，协程的一个概念。不做详细解释，可见简单理解为把每一条用例变成一个任务Task, 然后把任务放进一个列表中，最后以异步的方式，执行列表中的任务

class TestPlan(Test):def __init__(self):super().__init__()self.tasks = []def case_dispatch(self):  # create diff case_dispatch to yield diff dataframes from sheetsyield from ExcelHandler.read_TestCases(self.test_cases)def task_loader(self):  # load all the testcase to taskssem = asyncio.Semaphore(super().config['SEMAPHORE'])for case, test_id in self.case_dispatch():self.tasks.append(super().fetch_1(sem, case, test_id))

• case_dispatch 用做一个代理生成器。作用就是对一个用例dataframe中的内容不断产出， 当用例excel中又多个sheet，就会有多个dataframe, 代理生成器也可以设置多个，这样代码貌似便于理解。
• task_loader 加载任务到任务列表。目前在测试运行前就将任务全加载到列表中，效率和消耗较大。后期尝试是否可以改成动态加载。
• 执行

if __name__ == '__main__':st = time.time()plan = TestPlan()plan.task_loader()loop = asyncio.get_event_loop()future = asyncio.gather(*plan.tasks, return_exceptions=False)loop.run_until_complete(future)loop.close()ExcelHandler.save_testResult(plan.test_cases)print('timeUsage:', time.time() - st)

• 结果
  
  

未完待续。。。