test_rate_limit.py

# -*- coding: utf-8 -*-
"""
测试频率限制功能
"""
import time
import sys
import os

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 直接导入sunrequests模块
from adata.common.utils.sunrequests import sun_requests


def test_rate_limit():
    """测试频率限制"""
    print("测试频率限制功能...")
    
    # 测试同一个域名的请求
    url = "https://www.baidu.com"
    start_time = time.time()
    
    # 测试默认限制（30次/分钟）
    print("\n测试默认频率限制（30次/分钟）...")
    for i in range(35):  # 超过默认限制
        res = sun_requests.request('get', url)
        print(f"请求 {i+1}: 状态码={res.status_code}")
        if i == 29:  # 第30次请求后
            print(f"\n第30次请求完成，耗时: {time.time() - start_time:.2f}秒")
    
    end_time = time.time()
    print(f"\n35次请求总耗时: {end_time - start_time:.2f}秒")
    
    # 测试自定义频率限制
    print("\n\n测试自定义频率限制（10次/分钟）...")
    start_time = time.time()
    for i in range(15):  # 超过自定义限制
        res = sun_requests.request('get', url, rate_limit=10)
        print(f"请求 {i+1}: 状态码={res.status_code}")
        if i == 9:  # 第10次请求后
            print(f"\n第10次请求完成，耗时: {time.time() - start_time:.2f}秒")
    
    end_time = time.time()
    print(f"\n15次请求总耗时: {end_time - start_time:.2f}秒")
    
    print("\n测试完成！")


if __name__ == "__main__":
    test_rate_limit()