1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Path hack
import os, sys
try:
import unittest2 as unittest
except ImportError:
import unittest
from threading import Thread
from requests_cache import CachedSession
CACHE_NAME = 'requests_cache_test'
class ThreadSafetyTestCase(unittest.TestCase):
def test_caching_with_threads(self):
def do_tests_for(backend):
s = CachedSession(CACHE_NAME, backend)
s.cache.clear()
n_threads = 10
url = 'http://httpbin.org/get'
def do_requests(url, params):
for i in range(10): # for testing write and read from cache
s.get(url, params=params)
for _ in range(20): # stress test
threads = [Thread(target=do_requests, args=(url, {'param': i})) for i in range(n_threads)]
for t in threads:
t.start()
for t in threads:
t.join()
for i in range(n_threads):
self.assert_(s.cache.has_url('%s?param=%s' % (url, i)))
for backend in ('sqlite', 'mongodb'):
try:
do_tests_for(backend)
except Exception:
print("Failed to test %s" % backend)
if __name__ == '__main__':
unittest.main()
|