python周期性测试脚本

朱雀 2021-10-14 07:34 336阅读 0赞

import time
    import os
    import sched
    import threading
    import requests
    import unittest
    import json
    import traceback
    #定时任务案例
    
    # 初始化sched模块的scheduler类
    # 第一个参数是一个可以返回时间戳的函数，第二个参数可以在定时未到达之前阻塞。
    schedule = sched.scheduler(time.time, time.sleep)
    
    
    # 被周期性调度触发的函数
    def execute_command(cmd, inc):
        print('每三秒执行一次')
        url = ".../customer/getManager"
        headers = {"AccessToken": "1111"}
        querystring = {
            "mobile": 17277777777,
            "tenantId": "0d45d4de29be11e882b3d89d672b60b8"
        }
    
        try:
            # Post接口调用
            response = requests.request("POST", url, headers=headers, params=querystring)
            if response.status_code == 200:
                results = json.loads(response.text)
                if results['code'] == 1000:
                    print("Success")
                    print(results)
                else:
                    print("Fail")
                    print(results['data'])
            else:
                # 对于http返回非200的code，输出相应的code
                raise Exception("http error info:%s" % response.status_code)
        except:
            traceback.print_exc()
        # os.system(cmd)
        schedule.enter(inc, 0, execute_command, (cmd, inc))
    
    
    def main(cmd, inc=60):
        # enter四个参数分别为：间隔事件、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数，
        # 给该触发函数的参数（tuple形式）
        schedule.enter(0, 0, execute_command, (cmd, inc))
        schedule.run()
    
    
    # 每60秒查看下网络连接情况
    if __name__ == '__main__':
        main("dir", 3)
        # 每三秒执行  ""可传参数