test_ocsp.py 文件源码

def test_concurrent_ocsp_requests(tmpdir):
    from multiprocessing.pool import ThreadPool

    cache_file_name = path.join(str(tmpdir), 'cache_file.txt')
    urls = [
        'sfc-dev1-regression.s3.amazonaws.com',
        'sfctest0.snowflakecomputing.com',
        'sfc-ds2-customer-stage.s3.amazonaws.com',
        'snowflake.okta.com',
        'sfcdev1.blob.core.windows.net',
    ]
    ocsp_pyopenssl.OCSP_VALIDATION_CACHE = {}  # reset the memory cache
    urls = urls + urls + urls + urls + urls + urls
    pool = ThreadPool(len(urls))
    for url in urls:
        pool.apply_async(_validate_certs_using_ocsp, [url, cache_file_name])
    pool.close()
    pool.join()