Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
LOGGER.info(
'all {} modules chunks are set in uwsgi cache'.format(chunks))
uwsgi.cache_set('chunks-modules', repr(chunks), 0, cache_chunks)
chunks = int(math.ceil(len(json.dumps(vendors)) / float(64000)))
for i in range(0, chunks, 1):
uwsgi.cache_set('vendors-data{}'.format(i),
json.dumps(vendors)[i * 64000: (i + 1) * 64000],
0, main_cache)
LOGGER.info(
'all {} vendors chunks are set in uwsgi cache'.format(chunks))
uwsgi.cache_set('chunks-vendor', repr(chunks), 0, cache_chunks)
if response != 'work':
LOGGER.error('Could not load or create cache')
sys.exit(500)
uwsgi.cache_update('initialized', 'True', 0, cache_chunks)
def test_failed_by_one(self):
self.assertIsNone(uwsgi.cache_update('key1', 'HELLO', 0, 'items_1'))
def load(on_change):
"""Load all the data populated to yang-catalog to memory saved in file in ./cache."""
active_cache = get_active_cache()
if active_cache is None or on_change:
# We should get here only if application was started for the first time (active_cache is None)
# or if we need to reload cache (on_change == True)
with lock_for_load:
with lock_uwsgi_cache1:
LOGGER.info('Loading cache 1')
load_uwsgi_cache('cache_chunks1', 'main_cache1', 'cache_modules1', on_change)
# reset active cache back to 1 since we are done with populating cache 1
uwsgi.cache_update('active_cache', '1', 0, 'cache_chunks1')
LOGGER.info('Loading cache 2')
with lock_uwsgi_cache2:
load_uwsgi_cache('cache_chunks2', 'main_cache2', 'cache_modules2', on_change)
LOGGER.info('Both caches are loaded')
else:
# if we need to get some data from api
if active_cache[1] == '1':
# From cache 1
with lock_uwsgi_cache1:
load_uwsgi_cache('cache_chunks1', 'main_cache1', 'cache_modules1', on_change)
# reset active cache back to 1 since we are done with populating cache 1
uwsgi.cache_update('active_cache', '1', 0, 'cache_chunks1')
LOGGER.info('Using cache 1')
else:
with lock_uwsgi_cache2:
initialized = uwsgi.cache_get('initialized', 'cache_chunks2')
def test_big_item(self):
self.assertIsNone(uwsgi.cache_update('key1', 'HELLOHELLOHELLOHEL', 0, 'items_17'))
self.assertTrue(uwsgi.cache_update('key1', 'HELLOHELLOHELLOHE', 0, 'items_17'))
def test_overlapping(self):
self.assertTrue(uwsgi.cache_update('key1', 'HE', 0, 'items_2'))
self.assertIsNone(uwsgi.cache_update('key1', 'HELL', 0, 'items_2'))
self.assertTrue(uwsgi.cache_del('key1', 'items_2'))
self.assertTrue(uwsgi.cache_update('key1', 'HELL', 0, 'items_2'))
def on_post(self, req, resp):
body = load_body(req)
if uwsgi.cache_exists('configuration'):
uwsgi.cache_update('configuration', str(body))
else:
uwsgi.cache_set('configuration', str(body))
resp.status = falcon.HTTP_202
def test_two_items_using_four_blocks(self):
self.assertTrue(uwsgi.cache_update('key1', 'HE', 0, 'items_2'))
self.assertTrue(uwsgi.cache_update('key2', 'LL', 0, 'items_2'))
self.assertTrue(uwsgi.cache_del('key1', 'items_2'))
self.assertIsNone(uwsgi.cache_update('key1', 'HEL', 0, 'items_2'))
self.assertTrue(uwsgi.cache_update('key1', 'HE', 0, 'items_2'))
def test_overlapping(self):
self.assertTrue(uwsgi.cache_update('key1', 'HE', 0, 'items_2'))
self.assertIsNone(uwsgi.cache_update('key1', 'HELL', 0, 'items_2'))
self.assertTrue(uwsgi.cache_del('key1', 'items_2'))
self.assertTrue(uwsgi.cache_update('key1', 'HELL', 0, 'items_2'))
def test_big_update(self):
self.assertTrue(uwsgi.cache_set('key1', 'X' * 40, 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_update('key1', 'X' * 10, 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_del('key1', 'items_4_10'))
self.assertIsNone(uwsgi.cache_update('key1', 'X' * 51, 0, 'items_4_10'))
self.assertTrue(uwsgi.cache_update('key1', 'X' * 50, 0, 'items_4_10'))
LOGGER.info('Loading cache 1')
load_uwsgi_cache('cache_chunks1', 'main_cache1', 'cache_modules1', on_change)
# reset active cache back to 1 since we are done with populating cache 1
uwsgi.cache_update('active_cache', '1', 0, 'cache_chunks1')
LOGGER.info('Loading cache 2')
with lock_uwsgi_cache2:
load_uwsgi_cache('cache_chunks2', 'main_cache2', 'cache_modules2', on_change)
LOGGER.info('Both caches are loaded')
else:
# if we need to get some data from api
if active_cache[1] == '1':
# From cache 1
with lock_uwsgi_cache1:
load_uwsgi_cache('cache_chunks1', 'main_cache1', 'cache_modules1', on_change)
# reset active cache back to 1 since we are done with populating cache 1
uwsgi.cache_update('active_cache', '1', 0, 'cache_chunks1')
LOGGER.info('Using cache 1')
else:
with lock_uwsgi_cache2:
initialized = uwsgi.cache_get('initialized', 'cache_chunks2')
LOGGER.debug('initialized {} on change {}'.format(initialized, on_change))
if initialized is not None and initialized == 'True':
load_uwsgi_cache('cache_chunks2', 'main_cache2', 'cache_modules2', on_change)
LOGGER.info('Using cache 2')