46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import services.cache as cache_module
|
|
|
|
|
|
class RefreshableCacheDataVersionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.old_cache_dir = cache_module.CACHE_DIR
|
|
cache_module.CACHE_DIR = self.temp_dir.name
|
|
|
|
def tearDown(self):
|
|
cache_module.CACHE_DIR = self.old_cache_dir
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_data_version_change_invalidates_memory_and_file_cache(self):
|
|
source = mock.Mock(side_effect=[{'库存': 7642}, {'库存': 7566}])
|
|
cache = cache_module.RefreshableCache(maxsize=1, cache_key='test.inventory')
|
|
wrapped = cache.wrap(source)
|
|
|
|
self.assertEqual({'库存': 7642}, wrapped())
|
|
self.assertEqual({'库存': 7642}, wrapped())
|
|
self.assertEqual(1, source.call_count)
|
|
|
|
cache_module.bump_data_version()
|
|
|
|
self.assertEqual({'库存': 7566}, wrapped())
|
|
self.assertEqual(2, source.call_count)
|
|
|
|
def test_refresh_all_caches_publishes_new_data_version(self):
|
|
before = cache_module._data_version_signature()
|
|
cache_module.refresh_all_caches()
|
|
after = cache_module._data_version_signature()
|
|
|
|
self.assertIsNotNone(after)
|
|
self.assertNotEqual(before, after)
|
|
self.assertTrue(os.path.exists(
|
|
os.path.join(self.temp_dir.name, cache_module.DATA_VERSION_FILE)))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|