本文整理汇总了Python中zerver.lib.test_helpers.queries_captured函数的典型用法代码示例。如果您正苦于以下问题:Python queries_captured函数的具体用法?Python queries_captured怎么用?Python queries_captured使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了queries_captured函数的12个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_unread_count_user_soft_deactivation
def test_unread_count_user_soft_deactivation(self) -> None:
# In this test we make sure if a soft deactivated user had unread
# messages before deactivation they remain same way after activation.
long_term_idle_user = self.example_user('hamlet')
self.login(long_term_idle_user.email)
message = 'Test Message 1'
self.send_test_message(message)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 1)
query_count = len(queries)
user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(user_msg_list[-1].content, message)
self.logout()
do_soft_deactivate_users([long_term_idle_user])
self.login(long_term_idle_user.email)
message = 'Test Message 2'
self.send_test_message(message)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertNotEqual(idle_user_msg_list[-1].content, message)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 2)
# Test here for query count to be at least 5 greater than previous count
# This will assure indirectly that add_missing_messages() was called.
self.assertGreaterEqual(len(queries) - query_count, 5)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(idle_user_msg_list[-1].content, message)
开发者ID:umairwaheed,项目名称:zulip,代码行数:28,代码来源:test_home.py
示例2: test_multiple_user_soft_deactivations
def test_multiple_user_soft_deactivations(self):
# type: () -> None
long_term_idle_user = self.example_user('hamlet')
# We are sending this message to ensure that long_term_idle_user has
# at least one UserMessage row.
self.send_test_message('Testing', sender_name='hamlet')
do_soft_deactivate_users([long_term_idle_user])
message = 'Test Message 1'
self.send_test_message(message)
self.login(long_term_idle_user.email)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 2)
query_count = len(queries)
long_term_idle_user.refresh_from_db()
self.assertFalse(long_term_idle_user.long_term_idle)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(idle_user_msg_list[-1].content, message)
message = 'Test Message 2'
self.send_test_message(message)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 3)
# Test here for query count to be at least 5 less than previous count.
# This will assure add_missing_messages() isn't repeatedly called.
self.assertGreaterEqual(query_count - len(queries), 5)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(idle_user_msg_list[-1].content, message)
self.logout()
do_soft_deactivate_users([long_term_idle_user])
message = 'Test Message 3'
self.send_test_message(message)
self.login(long_term_idle_user.email)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 4)
query_count = len(queries)
long_term_idle_user.refresh_from_db()
self.assertFalse(long_term_idle_user.long_term_idle)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(idle_user_msg_list[-1].content, message)
message = 'Test Message 4'
self.send_test_message(message)
with queries_captured() as queries:
self.assertEqual(self.soft_activate_and_get_unread_count(), 5)
self.assertGreaterEqual(query_count - len(queries), 5)
idle_user_msg_list = get_user_messages(long_term_idle_user)
self.assertEqual(idle_user_msg_list[-1].content, message)
self.logout()
开发者ID:brockwhittaker,项目名称:zulip,代码行数:51,代码来源:test_home.py
示例3: test_num_queries_with_streams
def test_num_queries_with_streams(self) -> None:
main_user = self.example_user('hamlet')
other_user = self.example_user('cordelia')
realm_id = main_user.realm_id
self.login(main_user.email)
# Try to make page-load do extra work for various subscribed
# streams.
for i in range(10):
stream_name = 'test_stream_' + str(i)
stream = self.make_stream(stream_name)
DefaultStream.objects.create(
realm_id=realm_id,
stream_id=stream.id
)
for user in [main_user, other_user]:
self.subscribe(user, stream_name)
# Simulate hitting the page the first time to avoid some noise
# related to initial logins.
self._get_home_page()
# Then for the second page load, measure the number of queries.
flush_per_request_caches()
with queries_captured() as queries2:
result = self._get_home_page()
self.assert_length(queries2, 35)
# Do a sanity check that our new streams were in the payload.
html = result.content.decode('utf-8')
self.assertIn('test_stream_7', html)
开发者ID:umairwaheed,项目名称:zulip,代码行数:34,代码来源:test_home.py
示例4: test_json_users_with_bots
def test_json_users_with_bots(self) -> None:
hamlet = self.example_user('hamlet')
self.login(hamlet.email)
self.assert_num_bots_equal(0)
num_bots = 30
for i in range(num_bots):
full_name = 'Bot %d' % (i,)
short_name = 'bot-%d' % (i,)
bot_info = dict(
full_name=full_name,
short_name=short_name,
bot_type=1
)
result = self.client_post("/json/bots", bot_info)
self.assert_json_success(result)
self.assert_num_bots_equal(num_bots)
with queries_captured() as queries:
users_result = self.client_get('/json/users')
self.assert_json_success(users_result)
self.assert_length(queries, 4)
开发者ID:joydeep1701,项目名称:zulip,代码行数:25,代码来源:test_bots.py
示例5: test_num_queries_for_realm_admin
def test_num_queries_for_realm_admin(self) -> None:
# Verify number of queries for Realm admin isn't much higher than for normal users.
self.login(self.example_email("iago"))
flush_per_request_caches()
with queries_captured() as queries:
with patch('zerver.lib.cache.cache_set') as cache_mock:
result = self._get_home_page()
self.assertEqual(result.status_code, 200)
self.assert_length(cache_mock.call_args_list, 6)
self.assert_length(queries, 37)
开发者ID:umairwaheed,项目名称:zulip,代码行数:10,代码来源:test_home.py
示例6: test_activity
def test_activity(self) -> None:
self.login(self.example_email("hamlet"))
client, _ = Client.objects.get_or_create(name='website')
query = '/json/users/me/pointer'
last_visit = timezone_now()
count = 150
for activity_user_profile in UserProfile.objects.all():
UserActivity.objects.get_or_create(
user_profile=activity_user_profile,
client=client,
query=query,
count=count,
last_visit=last_visit
)
# Fails when not staff
result = self.client_get('/activity')
self.assertEqual(result.status_code, 302)
user_profile = self.example_user("hamlet")
user_profile.is_staff = True
user_profile.save()
flush_per_request_caches()
with queries_captured() as queries:
result = self.client_get('/activity')
self.assertEqual(result.status_code, 200)
self.assert_length(queries, 13)
flush_per_request_caches()
with queries_captured() as queries:
result = self.client_get('/realm_activity/zulip/')
self.assertEqual(result.status_code, 200)
self.assert_length(queries, 9)
flush_per_request_caches()
with queries_captured() as queries:
result = self.client_get('/user_activity/[email protected]/')
self.assertEqual(result.status_code, 200)
self.assert_length(queries, 5)
开发者ID:284928489,项目名称:zulip,代码行数:43,代码来源:test_presence.py
示例7: test_multiple_stream_senders
def test_multiple_stream_senders(self,
mock_send_future_email: mock.MagicMock,
mock_enough_traffic: mock.MagicMock) -> None:
client = 'website' # this makes `sent_by_human` return True
othello = self.example_user('othello')
self.subscribe(othello, 'Verona')
one_day_ago = timezone_now() - datetime.timedelta(days=1)
Message.objects.all().update(pub_date=one_day_ago)
one_sec_ago = timezone_now() - datetime.timedelta(seconds=1)
cutoff = time.mktime(one_sec_ago.timetuple())
senders = ['hamlet', 'cordelia', 'iago', 'prospero', 'ZOE']
for sender_name in senders:
email = self.example_email(sender_name)
self.login(email)
content = 'some content for ' + email
payload = dict(
type='stream',
client=client,
to='Verona',
topic='lunch',
content=content,
)
result = self.client_post("/json/messages", payload)
self.assert_json_success(result)
flush_per_request_caches()
with queries_captured() as queries:
handle_digest_email(othello.id, cutoff)
self.assertTrue(24 <= len(queries) <= 25)
self.assertEqual(mock_send_future_email.call_count, 1)
kwargs = mock_send_future_email.call_args[1]
self.assertEqual(kwargs['to_user_ids'], [othello.id])
hot_convo = kwargs['context']['hot_conversations'][0]
expected_participants = {
self.example_user(sender).full_name
for sender in senders
}
self.assertEqual(set(hot_convo['participants']), expected_participants)
self.assertEqual(hot_convo['count'], 5 - 2) # 5 messages, but 2 shown
teaser_messages = hot_convo['first_few_messages'][0]['senders']
self.assertIn('some content', teaser_messages[0]['content'][0]['plain'])
self.assertIn(teaser_messages[0]['sender'], expected_participants)
开发者ID:deltay,项目名称:zulip,代码行数:52,代码来源:test_digest.py
示例8: test_cache_behavior
def test_cache_behavior(self) -> None:
"""Tests whether fetching a user object the normal way, with
`get_user`, makes 1 cache query and 1 database query.
"""
realm = get_realm("zulip")
email = self.example_email("hamlet")
with queries_captured() as queries:
with simulated_empty_cache() as cache_queries:
user_profile = get_user(email, realm)
self.assert_length(queries, 1)
self.assert_length(cache_queries, 1)
self.assertEqual(user_profile.email, email)
开发者ID:rishig,项目名称:zulip,代码行数:13,代码来源:test_users.py
示例9: test_register
def test_register(self):
realm = get_realm("zulip.com")
streams = ["stream_%s" % i for i in range(40)]
for stream in streams:
create_stream_if_needed(realm, stream)
set_default_streams(realm, streams)
with queries_captured() as queries:
self.register("test", "test")
# Ensure the number of queries we make is not O(streams)
self.assert_length(queries, 74)
user_profile = get_user_profile_by_email('[email protected]')
self.assertEqual(get_session_dict_user(self.client.session), user_profile.id)
开发者ID:anindya,项目名称:zulip,代码行数:13,代码来源:test_signup.py
示例10: test_activity
def test_activity(self):
# type: () -> None
self.login("[email protected]")
client, _ = Client.objects.get_or_create(name='website')
query = '/json/users/me/pointer'
last_visit = datetime.datetime.now()
count=150
for user_profile in UserProfile.objects.all():
UserActivity.objects.get_or_create(
user_profile=user_profile,
client=client,
query=query,
count=count,
last_visit=last_visit
)
with queries_captured() as queries:
self.client_get('/activity')
self.assert_length(queries, 13)
开发者ID:tobby2002,项目名称:zulip,代码行数:19,代码来源:test_presence.py
示例11: test_activity
def test_activity(self) -> None:
self.login(self.example_email("hamlet"))
client, _ = Client.objects.get_or_create(name='website')
query = '/json/users/me/pointer'
last_visit = timezone_now()
count = 150
for user_profile in UserProfile.objects.all():
UserActivity.objects.get_or_create(
user_profile=user_profile,
client=client,
query=query,
count=count,
last_visit=last_visit
)
flush_per_request_caches()
with queries_captured() as queries:
self.client_get('/activity')
self.assert_length(queries, 4)
开发者ID:joydeep1701,项目名称:zulip,代码行数:19,代码来源:test_presence.py
示例12: test_home
#.........这里部分代码省略.........
"realm_inline_url_embed_preview",
"realm_invite_by_admins_only",
"realm_invite_required",
"realm_is_zephyr_mirror_realm",
"realm_mandatory_topics",
"realm_message_content_delete_limit_seconds",
"realm_message_content_edit_limit_seconds",
"realm_message_retention_days",
"realm_name",
"realm_name_changes_disabled",
"realm_name_in_notifications",
"realm_non_active_users",
"realm_notifications_stream_id",
"realm_password_auth_enabled",
"realm_presence_disabled",
"realm_push_notifications_enabled",
"realm_restricted_to_domain",
"realm_send_welcome_emails",
"realm_show_digest_email",
"realm_signup_notifications_stream_id",
"realm_uri",
"realm_user_groups",
"realm_users",
"realm_video_chat_provider",
"realm_waiting_period_threshold",
"root_domain_uri",
"save_stacktraces",
"server_generation",
"server_inline_image_preview",
"server_inline_url_embed_preview",
"stream_description_max_length",
"stream_name_max_length",
"subscriptions",
"test_suite",
"timezone",
"translate_emoticons",
"twenty_four_hour_time",
"unread_msgs",
"unsubscribed",
"use_websockets",
"user_id",
"warn_no_email",
"zulip_version",
]
email = self.example_email("hamlet")
# Verify fails if logged-out
result = self.client_get('/')
self.assertEqual(result.status_code, 302)
self.login(email)
# Create bot for realm_bots testing. Must be done before fetching home_page.
bot_info = {
'full_name': 'The Bot of Hamlet',
'short_name': 'hambot',
}
self.client_post("/json/bots", bot_info)
# Verify succeeds once logged-in
flush_per_request_caches()
with queries_captured() as queries:
with patch('zerver.lib.cache.cache_set') as cache_mock:
result = self._get_home_page(stream='Denmark')
self.assert_length(queries, 41)
self.assert_length(cache_mock.call_args_list, 7)
html = result.content.decode('utf-8')
for html_bit in html_bits:
if html_bit not in html:
raise AssertionError('%s not in result' % (html_bit,))
page_params = self._get_page_params(result)
actual_keys = sorted([str(k) for k in page_params.keys()])
self.assertEqual(actual_keys, expected_keys)
# TODO: Inspect the page_params data further.
# print(ujson.dumps(page_params, indent=2))
realm_bots_expected_keys = [
'api_key',
'avatar_url',
'bot_type',
'default_all_public_streams',
'default_events_register_stream',
'default_sending_stream',
'email',
'full_name',
'is_active',
'owner',
'services',
'user_id',
]
realm_bots_actual_keys = sorted([str(key) for key in page_params['realm_bots'][0].keys()])
self.assertEqual(realm_bots_actual_keys, realm_bots_expected_keys)
开发者ID:umairwaheed,项目名称:zulip,代码行数:101,代码来源:test_home.py
注:本文中的zerver.lib.test_helpers.queries_captured函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论