forked from NCAR/wrf-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcachetest.py
More file actions
68 lines (47 loc) · 1.51 KB
/
cachetest.py
File metadata and controls
68 lines (47 loc) · 1.51 KB
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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from __future__ import (absolute_import, division, print_function,
unicode_literals)
from threading import Thread
try:
from Queue import Queue
except ImportError:
from queue import Queue
from collections import OrderedDict
import unittest as ut
import numpy.testing as nt
from wrf.cache import cache_item, get_cached_item, _get_cache
from wrf.config import get_cache_size
class TestThread(Thread):
def __init__(self, num, q):
self.num = num
self.q = q
super(TestThread, self).__init__()
def run(self):
for i in range(get_cache_size() + 10):
key = "A" + str(i)
cache_item(key, "test", i * self.num)
item = get_cached_item(key, "test")
if item != i * self.num:
raise RuntimeError("cache is bogus")
cache = OrderedDict(_get_cache())
self.q.put(cache)
class CacheTest(ut.TestCase):
longMessage = True
def test_thread_local(self):
q1 = Queue()
q2 = Queue()
thread1 = TestThread(2, q1)
thread2 = TestThread(40, q2)
thread1.start()
thread2.start()
result1 = q1.get(True, 1)
result2 = q2.get(True, 1)
thread1.join()
thread2.join()
print(result1)
print(result2)
# Result 1 and 2 shoudl be different
self.assertNotEqual(result1, result2)
# This thread should have no cache
self.assertIsNone(_get_cache())
if __name__ == "__main__":
ut.main()