mirror of https://github.com/zulip/zulip.git
87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
from django.conf import settings
|
|
|
|
from mock import Mock, patch
|
|
from typing import List, Dict
|
|
|
|
from zerver.apps import flush_cache
|
|
from zerver.lib.cache import generic_bulk_cached_fetch, user_profile_by_email_cache_key
|
|
from zerver.lib.test_classes import ZulipTestCase
|
|
from zerver.models import get_system_bot, get_user_profile_by_email, UserProfile
|
|
|
|
class AppsTest(ZulipTestCase):
|
|
def test_cache_gets_flushed(self) -> None:
|
|
with patch('zerver.apps.logging.info') as mock_logging:
|
|
with patch('zerver.apps.cache.clear') as mock:
|
|
# The argument to flush_cache doesn't matter
|
|
flush_cache(Mock())
|
|
mock.assert_called_once()
|
|
mock_logging.assert_called_once()
|
|
|
|
class BotCacheKeyTest(ZulipTestCase):
|
|
def test_bot_profile_key_deleted_on_save(self) -> None:
|
|
# Get the profile cached on both cache keys:
|
|
user_profile = get_user_profile_by_email(settings.EMAIL_GATEWAY_BOT)
|
|
bot_profile = get_system_bot(settings.EMAIL_GATEWAY_BOT)
|
|
self.assertEqual(user_profile, bot_profile)
|
|
|
|
# Flip the setting and save:
|
|
flipped_setting = not bot_profile.is_api_super_user
|
|
bot_profile.is_api_super_user = flipped_setting
|
|
bot_profile.save()
|
|
|
|
# The .save() should have deleted cache keys, so if we fetch again,
|
|
# the returned objects should have is_api_super_user set correctly.
|
|
bot_profile2 = get_system_bot(settings.EMAIL_GATEWAY_BOT)
|
|
self.assertEqual(bot_profile2.is_api_super_user, flipped_setting)
|
|
|
|
user_profile2 = get_user_profile_by_email(settings.EMAIL_GATEWAY_BOT)
|
|
self.assertEqual(user_profile2.is_api_super_user, flipped_setting)
|
|
|
|
class GenericBulkCachedFetchTest(ZulipTestCase):
|
|
def test_query_function_called_only_if_needed(self) -> None:
|
|
# Get the user cached:
|
|
hamlet = get_user_profile_by_email(self.example_email("hamlet"))
|
|
|
|
class CustomException(Exception):
|
|
pass
|
|
|
|
def query_function(emails: List[str]) -> List[UserProfile]:
|
|
raise CustomException("The query function was called")
|
|
|
|
# query_function shouldn't be called, because the only requested object
|
|
# is already cached:
|
|
result = generic_bulk_cached_fetch(
|
|
cache_key_function=user_profile_by_email_cache_key,
|
|
query_function=query_function,
|
|
object_ids=[self.example_email("hamlet")]
|
|
) # type: Dict[str, UserProfile]
|
|
self.assertEqual(result, {hamlet.email: hamlet})
|
|
|
|
flush_cache(Mock())
|
|
# With the cache flushed, the query_function should get called:
|
|
with self.assertRaises(CustomException):
|
|
generic_bulk_cached_fetch(
|
|
cache_key_function=user_profile_by_email_cache_key,
|
|
query_function=query_function,
|
|
object_ids=[self.example_email("hamlet")]
|
|
)
|
|
|
|
def test_empty_object_ids_list(self) -> None:
|
|
class CustomException(Exception):
|
|
pass
|
|
|
|
def cache_key_function(email: str) -> str: # nocoverage -- this is just here to make sure it's not called
|
|
raise CustomException("The cache key function was called")
|
|
|
|
def query_function(emails: List[str]) -> List[UserProfile]: # nocoverage -- this is just here to make sure it's not called
|
|
raise CustomException("The query function was called")
|
|
|
|
# query_function and cache_key_function shouldn't be called, because
|
|
# objects_ids is empty, so there's nothing to do.
|
|
result = generic_bulk_cached_fetch(
|
|
cache_key_function=cache_key_function,
|
|
query_function=query_function,
|
|
object_ids=[]
|
|
) # type: Dict[str, UserProfile]
|
|
self.assertEqual(result, {})
|