File: test_thread_safety.py

package info (click to toggle)
python-requests-cache 0.4.13-1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 348 kB
  • ctags: 377
  • sloc: python: 1,359; makefile: 141
file content (48 lines) | stat: -rw-r--r-- 1,335 bytes parent folder | download | duplicates (3)
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()