wrf-python/test/cachetest.py

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()